app.py

import os
import subprocess
import datetime
import json
import time
import shutil
import uuid
import asyncio
import aiosqlite
from fastapi import FastAPI, UploadFile, File, BackgroundTasks, WebSocket, WebSocketDisconnect, Request, HTTPException
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from faster_whisper import WhisperModel
import requests
from collections import deque
import psutil
try:
    from dotenv import load_dotenv
    # Chargement des variables d'environnement (.env)
    load_dotenv()
except ImportError:
    print("DEBUG: python-dotenv non trouvé, utilisation des variables d'environnement système")

# Configuration des dossiers
UPLOAD_DIR = "uploads"
OUTPUT_DIR = "outputs"
STATIC_DIR = "static"
DB_PATH = "analytics.db"
os.makedirs(UPLOAD_DIR, exist_ok=True)
os.makedirs(OUTPUT_DIR, exist_ok=True)
os.makedirs(STATIC_DIR, exist_ok=True)

# --- CONFIGURATION OPTIMISATION ---
MAX_UPLOAD_SIZE = 5 * 1024 * 1024 * 1024  # 5GB max par fichier
CHUNK_SIZE = 10 * 1024 * 1024  # 10MB par chunk pour streaming
MAX_CONCURRENT_TASKS = 3  # Maximum de traitements simultanés
MAX_QUEUE_SIZE = 10  # Taille max de la file d'attente
MEMORY_THRESHOLD = 85  # Pourcentage de RAM max avant refus

# File d'attente pour gérer les traitements
processing_queue = deque()
active_tasks = set()
queue_lock = asyncio.Lock()

# --- BASE DE DONNÉES ---
async def init_db():
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS visitors (
                session_id TEXT PRIMARY KEY,
                start_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                last_activity TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        await db.execute("""
            CREATE TABLE IF NOT EXISTS feedback (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                client_id TEXT,
                rating TEXT,
                comment TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        await db.execute("""
            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                file_id TEXT UNIQUE,
                client_id TEXT,
                filename TEXT,
                status TEXT,
                progress INTEGER DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                result_url TEXT
            )
        """)
        await db.commit()

# --- NETTOYAGE AUTOMATIQUE ---
async def cleanup_old_files():
    """Supprime les fichiers de plus de 1 heure pour économiser l'espace disque"""
    while True:
        try:
            now = time.time()
            for folder in [UPLOAD_DIR, OUTPUT_DIR]:
                if not os.path.exists(folder): continue
                for filename in os.listdir(folder):
                    file_path = os.path.join(folder, filename)
                    if os.path.isfile(file_path):
                        if os.stat(file_path).st_mtime < now - 3600:
                            os.remove(file_path)
                            print(f"Cleanup: Suppression de {filename}")
        except Exception as e:
            print(f"Cleanup error: {e}")
        await asyncio.sleep(600) # Vérification toutes les 10 minutes

# --- GESTION DE LA FILE D'ATTENTE ---
async def check_system_resources():
    """Vérifie si le système a assez de ressources"""
    memory = psutil.virtual_memory()
    if memory.percent > MEMORY_THRESHOLD:
        return False, f"Mémoire saturée ({memory.percent}%)"
    return True, "OK"

async def process_queue():
    """Traite la file d'attente des tâches"""
    while True:
        try:
            async with queue_lock:
                # Notifier les utilisateurs en attente de leur position
                for idx, task_info in enumerate(processing_queue):
                    position = idx + 1
                    client_id = task_info.get('client_id')
                    file_id = task_info.get('file_id')
                    if client_id and file_id:
                        await send_status(
                            client_id,
                            f"En file d'attente (Position: {position}/{len(processing_queue)})",
                            5,
                            file_id
                        )

                # Si on a de la place et des tâches en attente
                if len(active_tasks) < MAX_CONCURRENT_TASKS and processing_queue:
                    task_info = processing_queue.popleft()
                    active_tasks.add(task_info['file_id'])

                    # Lancer le traitement en arrière-plan
                    asyncio.create_task(execute_task(task_info))
                    print(f"[QUEUE] Démarrage tâche {task_info['file_id']} ({len(active_tasks)}/{MAX_CONCURRENT_TASKS} actives)")
        except Exception as e:
            print(f"Queue error: {e}")

        await asyncio.sleep(2)  # Vérification toutes les 2 secondes

async def execute_task(task_info):
    """Exécute une tâche de traitement"""
    file_id = task_info['file_id']
    client_id = task_info['client_id']
    try:
        # Attendre un peu pour s'assurer que le WebSocket est connecté
        await asyncio.sleep(1)

        # Notifier que la tâche démarre
        await send_status(client_id, "Démarrage du traitement...", 10, file_id)
        await update_task_db(file_id, "En cours", 10)

        await run_hybrid_processing(
            file_id,
            client_id,
            task_info['audio_path']
        )
    except Exception as e:
        print(f"[TASK {file_id}] Erreur: {e}")
        await send_status(client_id, "Erreur de traitement", 0, file_id)
        await update_task_db(file_id, "Erreur", 0)
    finally:
        async with queue_lock:
            active_tasks.discard(file_id)
        print(f"[QUEUE] Tâche {file_id} terminée ({len(active_tasks)}/{MAX_CONCURRENT_TASKS} actives)")

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    cleanup_task = asyncio.create_task(cleanup_old_files())
    queue_task = asyncio.create_task(process_queue())
    yield
    cleanup_task.cancel()
    queue_task.cancel()

app = FastAPI(lifespan=lifespan)

async def track_visitor(session_id: str):
    try:
        if not session_id: return
        async with aiosqlite.connect(DB_PATH) as db:
            await db.execute("""
                INSERT INTO visitors (session_id) VALUES (?)
                ON CONFLICT(session_id) DO UPDATE SET last_activity = CURRENT_TIMESTAMP
            """, (session_id,))
            await db.commit()
    except Exception as e:
        print(f"Error tracking visitor: {e}")

async def update_task_db(file_id: str, status: str, progress: int, client_id: str = None, filename: str = None, result_url: str = None):
    try:
        async with aiosqlite.connect(DB_PATH) as db:
            if filename: # Nouvelle tâche
                await db.execute("""
                    INSERT INTO tasks (file_id, client_id, filename, status, progress)
                    VALUES (?, ?, ?, ?, ?)
                """, (file_id, client_id, filename, status, progress))
            else: # Mise à jour
                if result_url:
                    await db.execute("UPDATE tasks SET status = ?, progress = ?, result_url = ? WHERE file_id = ?", (status, progress, result_url, file_id))
                else:
                    await db.execute("UPDATE tasks SET status = ?, progress = ? WHERE file_id = ?", (status, progress, file_id))
            await db.commit()
    except Exception as e:
        print(f"Error updating task DB: {e}")

# Montage des fichiers statiques
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
app.mount("/outputs", StaticFiles(directory=OUTPUT_DIR), name="outputs")

# --- GLOBALS POUR SYNCHRONISATION HYBRIDE ---
# Stocke les segments corrigés en attendant que la vidéo finisse d'uploader
# { file_id: { "segments": [...], "video_ready": Event(), "video_path": str } }
hybrid_tasks = {}

async def run_hybrid_processing(file_id: str, client_id: str, audio_path: str):
    try:
        print(f"[{file_id}] Début du traitement hybride sur {audio_path}")
        await send_status(client_id, "Analyse audio accélérée...", 20, file_id)

        # 1. Transcription immédiate sur l'audio léger (Offload to thread to not block event loop)
        print(f"[{file_id}] Lancement Whisper...")
        def transcribe():
            segs, _ = model_whisper.transcribe(audio_path, word_timestamps=True, vad_filter=True, beam_size=1)
            return list(segs)

        segments = await asyncio.to_thread(transcribe)
        raw_texts = [s.text.strip() for s in segments]
        print(f"[{file_id}] Whisper terminé: {len(segments)} segments trouvés")

        # 2. Correction IA immédiate
        await send_status(client_id, "Correction IA (DeepSeek)...", 45, file_id)
        print(f"[{file_id}] Lancement DeepSeek...")
        corrected_texts = await deepseek_client.generate_correction(raw_texts)
        if corrected_texts is None:
            print(f"[{file_id}] DeepSeek a échoué ou n'a renvoyé aucun texte")
            corrected_texts = raw_texts
        print(f"[{file_id}] DeepSeek terminé")

        # Stockage des résultats intermédiaires
        if file_id not in hybrid_tasks:
            hybrid_tasks[file_id] = {"segments": None, "corrected": None, "video_ready": asyncio.Event(), "video_path": None}

        hybrid_tasks[file_id]["segments"] = segments
        hybrid_tasks[file_id]["corrected"] = corrected_texts

        await send_status(client_id, "IA terminée. Attente de la vidéo...", 65, file_id)
        print(f"[{file_id}] Phase IA ok, attente du signal video_ready...")

        # 3. Attente que la vidéo soit totalement uploadée (timeout de 10 min par sécurité)
        try:
            await asyncio.wait_for(hybrid_tasks[file_id]["video_ready"].wait(), timeout=600)
        except asyncio.TimeoutError:
            print(f"[{file_id}] Timeout: La vidéo n'a pas été reçue à temps")
            await send_status(client_id, "Erreur: Timeout upload vidéo", 0, file_id)
            return

        video_path = hybrid_tasks[file_id]["video_path"]
        print(f"[{file_id}] Vidéo reçue à {video_path}, lancement de la finalisation...")

        # 4. Finalisation (Stream + Fichier final)
        await send_status(client_id, "Vidéo prête !", 100, file_id)

        # On informe le client que le flux est prêt
        stream_url = f"/stream/{file_id}"
        if client_id in active_connections:
            await active_connections[client_id].send_json({
                "status": "Prêt !",
                "progress": 100,
                "stream_url": stream_url,
                "file_id": file_id
            })

        # On lance quand même la création du fichier final en tâche de fond pour le téléchargement
        asyncio.create_task(finalize_video(file_id, client_id, video_path, segments, corrected_texts))

    except Exception as e:
        import traceback
        traceback.print_exc()
        print(f"Erreur Hybrid [{file_id}]: {e}")
        await send_status(client_id, "Erreur de traitement", 0, file_id)
    finally:
        # Nettoyage
        if file_id in hybrid_tasks: del hybrid_tasks[file_id]
        if os.path.exists(audio_path):
            try: os.remove(audio_path)
            except: pass

async def finalize_video(file_id: str, client_id: str, video_path: str, segments, corrected_texts):
    try:
        output_video = os.path.join(OUTPUT_DIR, f"{file_id}_final.mp4")
        ass_path = os.path.join(UPLOAD_DIR, f"{file_id}.ass")

        await send_status(client_id, "Génération des styles...", 75, file_id)

        # On définit le header ici
        header = "[Script Info]\nScriptType: v4.00+\nPlayResX: 384\nPlayResY: 288\n\n[V4+ Styles]\nFormat: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding\nStyle: Premium,Arial,18,&H0000FFFF,&H00FFFFFF,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,2,0,2,10,10,50,1\n\n[Events]\nFormat: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\n"

        def write_ass():
            with open(ass_path, "w", encoding="utf-8") as f:
                f.write(header)
                for seg_orig, text_corr in zip(segments, corrected_texts):
                    words = text_corr.split()
                    if not words: continue
                    dur = (seg_orig.end - seg_orig.start) / max(1, len(words))
                    for i in range(0, len(words), 3):
                        chunk = words[i:i+3]
                        c_start = seg_orig.start + (i * dur)
                        c_end = c_start + (len(chunk) * dur)
                        line = "".join([f"{{\\k{max(1, int(dur*100))}}}{w.upper()} " for w in chunk])
                        f.write(f"Dialogue: 0,{format_ass_time(c_start)},{format_ass_time(c_end)},Premium,,0,0,0,,{line.strip()}\n")

        await asyncio.to_thread(write_ass)

        await send_status(client_id, "Incrustation finale...", 90, file_id)

        # FFmpeg peut prendre du temps, on l'exécute via asyncio pour ne pas bloquer
        cmd = ["ffmpeg", "-y", "-i", video_path, "-vf", f"ass={ass_path}", "-c:v", "libx264", "-preset", "ultrafast", "-crf", "22", "-c:a", "copy", "-threads", "0", output_video]

        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )

        stdout, stderr = await process.communicate()

        if process.returncode != 0:
            print(f"FFmpeg error: {stderr.decode()}")
            raise Exception("FFmpeg failed")

        result_url = f"/outputs/{file_id}_final.mp4"
        await send_status(client_id, "Prêt !", 100, file_id)
        await update_task_db(file_id, "Terminé", 100, result_url=result_url)

        if client_id in active_connections:
            await active_connections[client_id].send_json({"result_url": result_url})

    except Exception as e:
        print(f"Erreur Finalize: {e}")
        await send_status(client_id, "Erreur lors de l'incrustation", 0, file_id)

# --- LOGIQUE IA DEEPSEEK (PROMPT OPTIMISÉ) ---
DEEPSEEK_API_URL = "https://ds2api-tau-woad.vercel.app/v1/chat/completions"
DEEPSEEK_API_KEY = "sk-ds2api-key-1-your-custom-key"
DEEPSEEK_MODEL = "deepseek-chat"

class DeepSeekClient:
    def __init__(self):
        self.api_url = DEEPSEEK_API_URL
        self.api_key = DEEPSEEK_API_KEY
        self.model = DEEPSEEK_MODEL

    async def generate_correction(self, segments_text):
        if not self.api_key:
            print("DEBUG DEEPSEEK: API Key manquante")
            return None

        full_context = " ".join(segments_text)
        
        # PROMPT OPTIMISÉ - Version corrigée avec droits élargis
        prompt = (
            "Tu es un expert en post-production vidéo et correction de transcriptions (ASR).\n\n"
            "=== RÈGLE N°1 - SYNCHRO APPROXIMATIVE ===\n"
            "Chaque mot corrigé doit avoir une durée de prononciation PROCHE du mot d'origine.\n"
            "Tolérance autorisée : ±1 syllabe par mot. Si l'écart est plus grand, laisse l'original.\n\n"
            "=== RÈGLE N°2 - AJOUTS ET SUPPRESSIONS LIMITÉS ===\n"
            "Tu peux AJOUTER ou SUPPRIMER jusqu'à 2 mots par segment si nécessaire.\n"
            "Exceptions :\n"
            "- Tu peux ajouter 'ne' devant un verbe pour compléter une négation ('ne...pas')\n"
            "- Tu peux supprimer un mot parasite évident (ex: 'de' en trop, 'que' en trop)\n"
            "- Tu peux remplacer un mot inventé par le mot correct (même si longueur différente)\n\n"
            "=== RÈGLE N°3 - RÉÉCRITURE DE SEGMENT HALLUCINÉ ===\n"
            "Si dans un segment, PLUS DE 50% des mots sont des hallucinations ASR :\n"
            "- mots inexistants (ex: 'poisonnerait')\n"
            "- mots hors contexte total\n"
            "- séquence grammaticalement impossible\n"
            "ALORS tu peux RÉÉCRIRE LE SEGMENT ENTIÈREMENT en respectant le sens et la durée approximative.\n\n"
            "=== RÈGLE N°4 - COHÉRENCE GRAMMATICALE PRIORITAIRE ===\n"
            "Ces règles sont prioritaires sur le respect mot à mot :\n"
            "1. Sujet et verbe doivent s'accorder\n"
            "2. Les temps doivent être cohérents sur l'ensemble du texte\n"
            "3. Les pronoms (je/tu/il/on/nous/vous/ils) doivent être stables\n"
            "4. Les négations doivent être complètes ('ne...pas', 'ne...plus')\n\n"
            "=== RÈGLE N°5 - INTERDICTION D'HALLUCINER ===\n"
            "Si un mot est inaudible → garde-le tel quel.\n"
            "Si une correction est incertaine → garde l'original.\n"
            "Ne crée JAMAIS un mot qui n'existe pas dans la langue.\n"
            "Ne reformule PAS une phrase entière SAUF si la règle N°3 s'applique.\n\n"
            "=== CONTEXTE GLOBAL DE LA VIDÉO ===\n"
            f"\"{full_context}\"\n\n"
            "=== LISTE DES SEGMENTS À CORRIGER ===\n"
            f"{json.dumps(segments_text, ensure_ascii=False)}\n\n"
            "=== FORMAT DE RÉPONSE OBLIGATOIRE ===\n"
            "Renvoie UNIQUEMENT un JSON valide comme ceci :\n"
            "{\"corrected\": [\"segment1 corrigé\", \"segment2 corrigé\", ...]}\n\n"
            "=== RAPPEL FINAL ===\n"
            "Tu as le DROIT de :\n"
            "- ajouter/supprimer jusqu'à 2 mots par segment\n"
            "- réécrire entièrement un segment si plus de 50% est halluciné\n"
            "- modifier la longueur syllabique (tolérance ±1 syllabe)\n"
            "Tu ne dois PAS :\n"
            "- changer le nombre de segments\n"
            "- inventer des mots inexistants\n"
            "- inclure du texte en dehors du JSON"
        )

        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": "Tu es un assistant expert en correction de transcriptions. Tu réponds UNIQUEMENT en JSON valide. Tu as le droit d'ajouter/supprimer jusqu'à 2 mots par segment et de réécrire les segments hallucinés."},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.2,  # Légèrement augmenté pour permettre les corrections nécessaires
            "max_tokens": 4000
        }

        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }

        try:
            print(f"DEBUG DEEPSEEK: Envoi requête correction ({len(segments_text)} segments)...")
            resp = requests.post(self.api_url, headers=headers, json=payload, timeout=90)  # Timeout augmenté
            if resp.status_code == 200:
                data = resp.json()
                raw_text = data['choices'][0]['message']['content']
                clean_json = raw_text.replace('```json', '').replace('```', '').strip()
                result = json.loads(clean_json).get("corrected")
                
                # Vérification du nombre de segments (toujours obligatoire)
                if result and len(result) != len(segments_text):
                    print(f"ERREUR: Nombre de segments incorrect (attendu {len(segments_text)}, reçu {len(result)})")
                    return None
                    
                print(f"DEBUG DEEPSEEK: Correction réussie ({len(result) if result else 0} segments reçus)")
                return result
            else:
                print(f"DEBUG DEEPSEEK: Échec correction ({resp.status_code}): {resp.text}")
        except Exception as e:
            print(f"DEBUG DEEPSEEK: Exception lors de la correction: {e}")
        return None

@app.get("/stream/{file_id}")
async def stream_video(file_id: str):
    # On récupère les infos nécessaires
    # Note: Dans un vrai système, on utiliserait une DB ou un cache pour retrouver video_path et ass_path
    # Ici on cherche les fichiers basés sur l'ID
    video_files = [f for f in os.listdir(UPLOAD_DIR) if f.startswith(file_id + "_") and not f.endswith(".ass") and not f.endswith(".wav")]
    if not video_files:
        return JSONResponse(status_code=404, content={"error": "Video not found"})

    video_path = os.path.join(UPLOAD_DIR, video_files[0])
    ass_path = os.path.join(UPLOAD_DIR, f"{file_id}.ass")

    if not os.path.exists(ass_path):
        return JSONResponse(status_code=404, content={"error": "Subtitles not ready"})

    async def video_generator():
        cmd = [
            "ffmpeg", "-y", "-i", video_path,
            "-vf", f"ass={ass_path}",
            "-c:v", "libx264", "-preset", "ultrafast", "-crf", "22",
            "-c:a", "copy",
            "-f", "mp4",
            "-movflags", "frag_keyframe+empty_moov+default_base_moof",
            "pipe:1"
        ]

        process = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )

        try:
            while True:
                data = await process.stdout.read(1024 * 64) # 64KB chunks
                if not data:
                    break
                yield data
        except Exception as e:
            print(f"Streaming error: {e}")
        finally:
            try:
                process.terminate()
            except:
                pass

    return StreamingResponse(video_generator(), media_type="video/mp4")

# --- ENDPOINTS FEEDBACK ---
@app.post("/api/feedback")
async def post_feedback(request: Request):
    try:
        data = await request.json()
        client_id = data.get("client_id")
        rating = data.get("rating")
        comment = data.get("comment")

        async with aiosqlite.connect(DB_PATH) as db:
            await db.execute("""
                INSERT INTO feedback (client_id, rating, comment)
                VALUES (?, ?, ?)
            """, (client_id, rating, comment))
            await db.commit()
        return {"status": "success"}
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})

@app.get("/api/feedbacks")
async def get_feedbacks():
    try:
        async with aiosqlite.connect(DB_PATH) as db:
            db.row_factory = aiosqlite.Row
            cursor = await db.execute("SELECT * FROM feedback ORDER BY created_at DESC")
            rows = await cursor.fetchall()
            return [dict(row) for row in rows]
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": str(e)})

deepseek_client = DeepSeekClient()
# On utilise local_files_only=False par défaut mais on a déjà pré-téléchargé dans le Dockerfile
model_whisper = WhisperModel("base", device="cpu", compute_type="int8", cpu_threads=4)

active_connections = {}

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await websocket.accept()
    await track_visitor(client_id)
    active_connections[client_id] = websocket
    try:
        while True:
            data = await websocket.receive_text()
            await track_visitor(client_id) # Update activity
    except WebSocketDisconnect:
        if client_id in active_connections: del active_connections[client_id]

async def send_status(client_id: str, status: str, progress: int, file_id: str = None):
    if file_id:
        await update_task_db(file_id, status, progress)

    if client_id in active_connections:
        try:
            await active_connections[client_id].send_json({
                "status": status,
                "progress": progress,
                "file_id": file_id
            })
            print(f"[WS] Envoyé à {client_id}: {status} ({progress}%)")
        except Exception as e:
            print(f"[WS] Erreur envoi à {client_id}: {e}")
    else:
        print(f"[WS] Client {client_id} non connecté, message ignoré: {status}")

def format_ass_time(seconds: float):
    td = datetime.timedelta(seconds=seconds)
    total_sec = int(td.total_seconds())
    h, m, s = total_sec // 3600, (total_sec % 3600) // 60, total_sec % 60
    cs = int((seconds - int(seconds)) * 100)
    return f"{h}:{m:02d}:{s:02d}.{cs:02d}"

async def run_processing(video_path: str, client_id: str, file_id: str):
    try:
        output_video = os.path.join(OUTPUT_DIR, f"{file_id}_final.mp4")
        ass_path = os.path.join(UPLOAD_DIR, f"{file_id}.ass")

        await send_status(client_id, "Analyse audio...", 20, file_id)
        segments, _ = model_whisper.transcribe(video_path, word_timestamps=True, vad_filter=True, beam_size=1)
        segments = list(segments)
        raw_texts = [s.text.strip() for s in segments]

        await send_status(client_id, "Correction IA (DeepSeek)...", 45, file_id)
        corrected_texts = await deepseek_client.generate_correction(raw_texts)
        if corrected_texts is None: corrected_texts = raw_texts

        await send_status(client_id, "Génération des styles...", 70, file_id)
        # Style Premium : Jaune vibrant, bordure noire, police Arial grasse
        header = "[Script Info]\nScriptType: v4.00+\nPlayResX: 384\nPlayResY: 288\n\n[V4+ Styles]\nFormat: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding\nStyle: Premium,Arial,18,&H0000FFFF,&H00FFFFFF,&H00000000,&H00000000,-1,0,0,0,100,100,0,0,1,2,0,2,10,10,50,1\n\n[Events]\nFormat: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\n"

        with open(ass_path, "w", encoding="utf-8") as f:
            f.write(header)
            for seg_orig, text_corr in zip(segments, corrected_texts):
                words = text_corr.split()
                if not words: continue
                dur = (seg_orig.end - seg_orig.start) / max(1, len(words))
                for i in range(0, len(words), 3):
                    chunk = words[i:i+3]
                    c_start = seg_orig.start + (i * dur)
                    c_end = c_start + (len(chunk) * dur)
                    # Animation Karaoké (\k) pour l'effet dynamique
                    line = "".join([f"{{\\k{max(1, int(dur*100))}}}{w.upper()} " for w in chunk])
                    f.write(f"Dialogue: 0,{format_ass_time(c_start)},{format_ass_time(c_end)},Premium,,0,0,0,,{line.strip()}\n")

        await send_status(client_id, "Finalisation vidéo...", 85, file_id)
        # Encodage ultra-rapide
        subprocess.run(["ffmpeg", "-y", "-i", video_path, "-vf", f"ass={ass_path}", "-c:v", "libx264", "-preset", "ultrafast", "-crf", "22", "-c:a", "copy", "-threads", "0", output_video], check=True)

        result_url = f"/outputs/{file_id}_final.mp4"
        await send_status(client_id, "Prêt !", 100, file_id)
        await update_task_db(file_id, "Terminé", 100, result_url=result_url)

        if client_id in active_connections:
            await active_connections[client_id].send_json({"result_url": result_url})

    except Exception as e:
        print(f"Erreur : {e}")
        await send_status(client_id, f"Erreur de traitement", 0, file_id)
        await update_task_db(file_id, "Erreur", 0)

@app.get("/")
async def read_index():
    return FileResponse(os.path.join(STATIC_DIR, "index.html"))

import re

def sanitize_filename(filename: str) -> str:
    # Ne garde que les caractères alphanumériques, points, tirets et underscores
    return re.sub(r'[^a-zA-Z0-9._-]', '_', filename)

@app.post("/upload")
async def upload_video(background_tasks: BackgroundTasks, file: UploadFile = File(...), client_id: str = None, file_id: str = None, type: str = "video"):
    # Vérification des ressources système
    resources_ok, msg = await check_system_resources()
    if not resources_ok:
        raise HTTPException(status_code=503, detail=f"Serveur surchargé: {msg}")

    # Vérification de la file d'attente
    async with queue_lock:
        queue_size = len(processing_queue) + len(active_tasks)
        if queue_size >= MAX_QUEUE_SIZE:
            raise HTTPException(status_code=429, detail=f"Trop de requêtes en cours. Réessayez dans quelques instants. ({queue_size}/{MAX_QUEUE_SIZE})")

    # Sécurisation : on génère un nouvel UUID si celui fourni est suspect ou absent
    safe_file_id = str(uuid.uuid4())
    if file_id and re.match(r'^[a-f0-9-]{36}$', file_id):
        safe_file_id = file_id

    safe_filename = sanitize_filename(file.filename)

    if type == "audio":
        # Phase 1 : L'audio léger arrive en premier - STREAMING OPTIMISÉ
        audio_path = os.path.join(UPLOAD_DIR, f"{safe_file_id}_temp.wav")

        # Upload par chunks pour éviter de saturer la mémoire
        total_size = 0
        try:
            async with asyncio.Lock():  # Éviter les conflits d'écriture
                with open(audio_path, "wb") as buffer:
                    while chunk := await file.read(CHUNK_SIZE):
                        total_size += len(chunk)
                        if total_size > MAX_UPLOAD_SIZE:
                            os.remove(audio_path)
                            raise HTTPException(status_code=413, detail=f"Fichier trop volumineux (max {MAX_UPLOAD_SIZE // (1024**3)}GB)")
                        buffer.write(chunk)
                        await asyncio.sleep(0)  # Permet à d'autres tâches de s'exécuter
        except Exception as e:
            if os.path.exists(audio_path):
                os.remove(audio_path)
            raise HTTPException(status_code=500, detail=f"Erreur upload: {str(e)}")

        # Enregistrer la tâche
        await update_task_db(safe_file_id, "En attente", 5, client_id, safe_filename)
        if client_id: await track_visitor(client_id)

        # Ajouter à la file d'attente au lieu de lancer directement
        async with queue_lock:
            processing_queue.append({
                'file_id': safe_file_id,
                'client_id': client_id,
                'audio_path': audio_path
            })
            queue_position = len(processing_queue)

        print(f"[QUEUE] Tâche {safe_file_id} ajoutée (position: {queue_position})")
        return {"file_id": safe_file_id, "status": "audio_received", "queue_position": queue_position}

    else:
        # Phase 2 : La vidéo complète arrive - STREAMING OPTIMISÉ
        video_path = os.path.join(UPLOAD_DIR, f"{safe_file_id}_{safe_filename}")

        total_size = 0
        try:
            async with asyncio.Lock():
                with open(video_path, "wb") as buffer:
                    while chunk := await file.read(CHUNK_SIZE):
                        total_size += len(chunk)
                        if total_size > MAX_UPLOAD_SIZE:
                            os.remove(video_path)
                            raise HTTPException(status_code=413, detail=f"Fichier trop volumineux (max {MAX_UPLOAD_SIZE // (1024**3)}GB)")
                        buffer.write(chunk)
                        await asyncio.sleep(0)
        except Exception as e:
            if os.path.exists(video_path):
                os.remove(video_path)
            raise HTTPException(status_code=500, detail=f"Erreur upload: {str(e)}")

        # Notifier que la vidéo est là
        if safe_file_id not in hybrid_tasks:
            hybrid_tasks[safe_file_id] = {"segments": None, "corrected": None, "video_ready": asyncio.Event(), "video_path": None}

        hybrid_tasks[safe_file_id]["video_path"] = video_path
        hybrid_tasks[safe_file_id]["video_ready"].set()

        return {"file_id": safe_file_id, "status": "video_received"}

# --- DASHBOARD API ---
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin123")

@app.get("/dashboard")
async def get_dashboard(password: str = None):
    if password != ADMIN_PASSWORD:
        return JSONResponse(status_code=403, content={"error": "Accès refusé. Ajoutez ?password=VOTRE_MOT_DE_PASSE à l'URL"})
    return FileResponse(os.path.join(STATIC_DIR, "dashboard.html"))

@app.get("/api/stats")
async def get_stats(password: str = None):
    if password != ADMIN_PASSWORD:
        return JSONResponse(status_code=403, content={"error": "Accès refusé"})
    try:
        async with aiosqlite.connect(DB_PATH) as db:
            db.row_factory = aiosqlite.Row

            cursor = await db.execute("SELECT COUNT(*) as count FROM visitors")
            total_visitors = (await cursor.fetchone())['count']

            cursor = await db.execute("SELECT COUNT(*) as count FROM tasks")
            total_tasks = (await cursor.fetchone())['count']

            cursor = await db.execute("SELECT COUNT(*) as count FROM tasks WHERE status = 'Terminé'")
            success_tasks = (await cursor.fetchone())['count']

            cursor = await db.execute("SELECT AVG(strftime('%s', last_activity) - strftime('%s', start_time)) as avg_dur FROM visitors")
            avg_duration = (await cursor.fetchone())['avg_dur'] or 0

            return {
                "active_users": len(active_connections),
                "total_visitors": total_visitors,
                "total_tasks": total_tasks,
                "success_rate": round((success_tasks / max(1, total_tasks) * 100), 1),
                "avg_session_seconds": round(avg_duration),
                "queue_size": len(processing_queue),
                "active_tasks": len(active_tasks),
                "memory_usage": psutil.virtual_memory().percent
            }
    except Exception as e:
        return {"error": str(e)}

@app.get("/api/history")
async def get_history(password: str = None):
    if password != ADMIN_PASSWORD:
        return JSONResponse(status_code=403, content={"error": "Accès refusé"})
    try:
        async with aiosqlite.connect(DB_PATH) as db:
            db.row_factory = aiosqlite.Row
            cursor = await db.execute("SELECT * FROM tasks ORDER BY created_at DESC LIMIT 50")
            rows = await cursor.fetchall()
            return [dict(row) for row in rows]
    except Exception as e:
        return {"error": str(e)}

@app.get("/api/queue-status")
async def get_queue_status(client_id: str = None):
    """Endpoint pour vérifier le statut de la file d'attente"""
    async with queue_lock:
        queue_info = {
            "queue_size": len(processing_queue),
            "active_tasks": len(active_tasks),
            "max_concurrent": MAX_CONCURRENT_TASKS,
            "max_queue": MAX_QUEUE_SIZE,
            "available_slots": MAX_QUEUE_SIZE - (len(processing_queue) + len(active_tasks)),
            "memory_usage": psutil.virtual_memory().percent,
            "system_ready": psutil.virtual_memory().percent < MEMORY_THRESHOLD
        }

        # Si un client_id est fourni, chercher sa position dans la queue
        if client_id:
            position = None
            for idx, task in enumerate(processing_queue):
                if task['client_id'] == client_id:
                    position = idx + 1
                    break
            queue_info["your_position"] = position
            queue_info["is_processing"] = any(task['client_id'] == client_id for task in processing_queue if task['file_id'] in active_tasks)

        return queue_info

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)