app.py

"""
DubFlow Web v2.4 — Real-time + REST API + Fixed Pipe Handling
"""

from fastapi import FastAPI, UploadFile, File, HTTPException, Request, Query, BackgroundTasks
from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware

import asyncio, subprocess, uuid, os, re, json, time, shutil, hashlib, logging
from collections import OrderedDict
from contextlib import asynccontextmanager
from concurrent.futures import ThreadPoolExecutor
from typing import Optional

import aiohttp, aiofiles, httpx, edge_tts

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
log = logging.getLogger("dubflow")

TMP = "/tmp"
PUBLIC = "/tmp/public"
os.makedirs(PUBLIC, exist_ok=True)

FFMPEG = shutil.which("ffmpeg") or "ffmpeg"
FFPROBE = shutil.which("ffprobe") or "ffprobe"

TTS_CONCURRENCY = 12
TRANS_CONCURRENCY = 20
FFMPEG_THREADS = 6
CHUNK_SIZE = 1024 * 512

MAX_UPLOAD_MB = 500
MIN_FILE_BYTES = 1000
FILE_TTL_SECONDS = 3600

STEP_WEIGHTS = {0: 3, 1: 35, 2: 12, 3: 35, 4: 15}

JOBS: dict = {}
THREAD_POOL = ThreadPoolExecutor(max_workers=32)


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(periodic_cleanup())
    log.info("DubFlow v2.6.0 started")
    yield
    task.cancel()
    THREAD_POOL.shutdown(wait=False)


app = FastAPI(
    title="DubFlow API",
    version="2.6.0",
    description="Video dubbing pipeline — MP4 → SRT → Translate → TTS → Dubbed MP4",
    lifespan=lifespan,
)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
app.add_middleware(GZipMiddleware, minimum_size=1000)


async def periodic_cleanup():
    while True:
        try:
            await asyncio.sleep(600)
            now = time.time()
            for fn in os.listdir(PUBLIC):
                p = os.path.join(PUBLIC, fn)
                try:
                    if now - os.path.getmtime(p) > FILE_TTL_SECONDS:
                        os.remove(p)
                except: pass
            dead = [k for k, v in JOBS.items()
                    if v.get("done") and now - v.get("created", now) > FILE_TTL_SECONDS]
            for k in dead: JOBS.pop(k, None)
        except asyncio.CancelledError: break
        except Exception as e: log.error(f"cleanup: {e}")


# ══════════════════════════════════════════════════════════════
# VOICES & LANGUAGES
# ══════════════════════════════════════════════════════════════
VOICES = {
    ("af","Female"):"af-ZA-AdriNeural", ("af","Male"):"af-ZA-WillemNeural",
    ("sq","Female"):"sq-AL-AnilaNeural", ("sq","Male"):"sq-AL-IlirNeural",
    ("ar","Female"):"ar-SA-ZariyahNeural", ("ar","Male"):"ar-SA-HamedNeural",
    ("bg","Female"):"bg-BG-KalinaNeural", ("bg","Male"):"bg-BG-BorislavNeural",
    ("zh","Female"):"zh-CN-XiaoxiaoNeural", ("zh","Male"):"zh-CN-YunxiNeural",
    ("hr","Female"):"hr-HR-GabrijelaNeural", ("hr","Male"):"hr-HR-SreckoNeural",
    ("cs","Female"):"cs-CZ-VlastaNeural", ("cs","Male"):"cs-CZ-AntoninNeural",
    ("da","Female"):"da-DK-ChristelNeural", ("da","Male"):"da-DK-JeppeNeural",
    ("nl","Female"):"nl-NL-ColetteNeural", ("nl","Male"):"nl-NL-MaartenNeural",
    ("en","Female"):"en-US-AriaNeural", ("en","Male"):"en-US-GuyNeural",
    ("fi","Female"):"fi-FI-NooraNeural", ("fi","Male"):"fi-FI-HarriNeural",
    ("fr","Female"):"fr-FR-DeniseNeural", ("fr","Male"):"fr-FR-HenriNeural",
    ("de","Female"):"de-DE-AmalaNeural", ("de","Male"):"de-DE-ConradNeural",
    ("el","Female"):"el-GR-AthinaNeural", ("el","Male"):"el-GR-NestorasNeural",
    ("he","Female"):"he-IL-HilaNeural", ("he","Male"):"he-IL-AvriNeural",
    ("hi","Female"):"hi-IN-SwaraNeural", ("hi","Male"):"hi-IN-MadhurNeural",
    ("hu","Female"):"hu-HU-NoemiNeural", ("hu","Male"):"hu-HU-TamasNeural",
    ("id","Female"):"id-ID-GadisNeural", ("id","Male"):"id-ID-ArdiNeural",
    ("it","Female"):"it-IT-IsabellaNeural", ("it","Male"):"it-IT-DiegoNeural",
    ("ja","Female"):"ja-JP-NanamiNeural", ("ja","Male"):"ja-JP-KeitaNeural",
    ("ko","Female"):"ko-KR-SunHiNeural", ("ko","Male"):"ko-KR-InJoonNeural",
    ("ms","Female"):"ms-MY-YasminNeural", ("ms","Male"):"ms-MY-OsmanNeural",
    ("nb","Female"):"nb-NO-PernilleNeural", ("nb","Male"):"nb-NO-FinnNeural",
    ("fa","Female"):"fa-IR-DilaraNeural", ("fa","Male"):"fa-IR-FaridNeural",
    ("pl","Female"):"pl-PL-ZofiaNeural", ("pl","Male"):"pl-PL-MarekNeural",
    ("pt","Female"):"pt-BR-FranciscaNeural", ("pt","Male"):"pt-BR-AntonioNeural",
    ("ro","Female"):"ro-RO-AlinaNeural", ("ro","Male"):"ro-RO-EmilNeural",
    ("ru","Female"):"ru-RU-SvetlanaNeural", ("ru","Male"):"ru-RU-DmitryNeural",
    ("sk","Female"):"sk-SK-ViktoriaNeural", ("sk","Male"):"sk-SK-LukasNeural",
    ("es","Female"):"es-ES-ElviraNeural", ("es","Male"):"es-ES-AlvaroNeural",
    ("sv","Female"):"sv-SE-SofieNeural", ("sv","Male"):"sv-SE-MattiasNeural",
    ("th","Female"):"th-TH-PremwadeeNeural", ("th","Male"):"th-TH-NiwatNeural",
    ("tr","Female"):"tr-TR-EmelNeural", ("tr","Male"):"tr-TR-AhmetNeural",
    ("uk","Female"):"uk-UA-PolinaNeural", ("uk","Male"):"uk-UA-OstapNeural",
    ("vi","Female"):"vi-VN-HoaiMyNeural", ("vi","Male"):"vi-VN-NamMinhNeural",
}

LANGUAGES = OrderedDict([
    ("it","Italiano"),("en","English"),("es","Español"),("fr","Français"),
    ("de","Deutsch"),("pt","Português"),("ru","Русский"),("zh","中文"),
    ("ja","日本語"),("ko","한국어"),("ar","العربية"),("hi","हिन्दी"),
    ("tr","Türkçe"),("pl","Polski"),("nl","Nederlands"),("sv","Svenska"),
    ("da","Dansk"),("fi","Suomi"),("nb","Norsk"),("el","Ελληνικά"),
    ("cs","Čeština"),("ro","Română"),("hu","Magyar"),("uk","Українська"),
    ("id","Indonesia"),("th","ไทย"),("vi","Tiếng Việt"),("he","עברית"),
    ("bg","Български"),("hr","Hrvatski"),("sk","Slovenčina"),("sq","Shqip"),
    ("af","Afrikaans"),("ms","Melayu"),("fa","فارسی"),
])


# ══════════════════════════════════════════════════════════════
# JOB MANAGEMENT
# ══════════════════════════════════════════════════════════════
def new_job():
    jid = uuid.uuid4().hex
    JOBS[jid] = {
        "stage": "init", "message": "In coda",
        "step": 0, "step_progress": 0.0, "progress": 0.0,
        "result": None, "done": False, "error": None,
        "cancel": False, "created": time.time(),
        "video_ready": False, "video_path": None, "params": None,
        "last_heartbeat": time.time(),
        "original_name": "video",
    }
    log.info(f"[{jid[:8]}] job created")
    return jid

def compute_global(step, step_pct):
    """Calcolo progress globale con precisione decimale."""
    done = sum(STEP_WEIGHTS[s] for s in STEP_WEIGHTS if s < step)
    if step in STEP_WEIGHTS:
        done += STEP_WEIGHTS[step] * (step_pct / 100.0)
    return round(min(100.0, done), 2)

def update(jid, **kw):
    j = JOBS.get(jid)
    if not j: return
    j.update(kw)
    if "step" in kw or "step_progress" in kw:
        j["progress"] = round(compute_global(j["step"], j["step_progress"]), 2)
    j["last_heartbeat"] = time.time()

def is_cancelled(jid):
    return JOBS.get(jid, {}).get("cancel", False)

def cleanup_paths(paths):
    for p in paths:
        try:
            if not p: continue
            if os.path.isdir(p): shutil.rmtree(p, ignore_errors=True)
            elif os.path.isfile(p): os.remove(p)
        except: pass


# ══════════════════════════════════════════════════════════════
# SRT UTILS
# ══════════════════════════════════════════════════════════════
_TS_RE = re.compile(r'(\d{1,2}:\d{2}:\d{2}[.,]\d{3})\s*-->\s*(\d{1,2}:\d{2}:\d{2}[.,]\d{3})')
_BLANK_RE = re.compile(r'\n\s*\n')
_IDX_RE = re.compile(r'^\d+\s*$')

def ts_to_sec(ts):
    ts = ts.strip().replace(",", ".")
    parts = ts.split(":")
    if len(parts) == 3:
        return int(parts[0])*3600 + int(parts[1])*60 + float(parts[2])
    if len(parts) == 2:
        return int(parts[0])*60 + float(parts[1])
    return float(ts)

def parse_srt(content):
    blocks = []
    content = content.replace('\r\n','\n').replace('\r','\n').lstrip('\ufeff')
    for raw in _BLANK_RE.split(content.strip()):
        lines = raw.strip().split('\n')
        if len(lines) < 2 or not _IDX_RE.match(lines[0].strip()):
            continue
        ts_m = _TS_RE.match(lines[1].strip())
        if not ts_m: continue
        text = '\n'.join(lines[2:]).strip()
        if text:
            blocks.append({
                "index": int(lines[0].strip()),
                "timestamp": lines[1].strip(),
                "start_sec": ts_to_sec(ts_m.group(1)),
                "end_sec": ts_to_sec(ts_m.group(2)),
                "text": text,
            })
    return blocks

def build_srt(blocks):
    return '\n'.join(f"{i}\n{b['timestamp']}\n{b['text']}\n" for i, b in enumerate(blocks, 1))

_CLEAN_PATTERNS = [
    (re.compile(r'<[^>]+>'), ''),
    (re.compile(r'\{[^}]+\}'), ''),
    (re.compile(r'♪[^♪]*♪'), ''), (re.compile(r'♪'), ''),
    (re.compile(r'\[.*?\]'), ''), (re.compile(r'\(.*?\)'), ''),
    (re.compile(r'-{2,}'), ''), (re.compile(r'\s+'), ' '),
]

def clean_text(text):
    for pat, rep in _CLEAN_PATTERNS:
        text = pat.sub(rep, text)
    return text.strip()


# ══════════════════════════════════════════════════════════════
# FFMPEG HELPERS (FIXED — no pipe race)
# ══════════════════════════════════════════════════════════════
async def get_video_duration_async(path):
    try:
        proc = await asyncio.create_subprocess_exec(
            FFPROBE, "-v", "error", "-show_entries", "format=duration",
            "-of", "default=noprint_wrappers=1:nokey=1", path,
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
        out, _ = await proc.communicate()
        return float(out.decode().strip() or 0)
    except: return 0.0

async def validate_video(path):
    cmd = [FFPROBE, "-v", "error", "-select_streams", "v:0",
           "-show_entries", "stream=codec_name,duration", "-of", "json", path]
    proc = await asyncio.create_subprocess_exec(
        *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    out, err = await proc.communicate()
    if proc.returncode != 0:
        raise RuntimeError(f"Video non valido: {err.decode()[:200]}")
    try:
        info = json.loads(out.decode() or "{}")
        if not info.get("streams"):
            raise RuntimeError("Nessun flusso video")
    except json.JSONDecodeError:
        raise RuntimeError("Output ffprobe non valido")


async def run_ffmpeg_with_progress(cmd, duration, jid, base_pct, span_pct, msg_prefix=""):
    """
    Esegue ffmpeg con -progress pipe:1 leggendo stdout/stderr in parallelo.
    Evita race condition di communicate() + manual read.
    """
    proc = await asyncio.create_subprocess_exec(
        *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    
    stderr_buf = bytearray()
    
    async def read_stdout():
        try:
            while True:
                line = await proc.stdout.readline()
                if not line: break
                s = line.decode(errors="ignore").strip()
                if s.startswith("out_time_ms=") and duration > 0:
                    try:
                        us = int(s.split("=", 1)[1])
                        sec = us / 1_000_000
                        pct = base_pct + min(span_pct, (sec/duration) * span_pct)
                        update(jid, step_progress=round(pct, 2),
                               message=f"{msg_prefix}{sec:.1f}/{duration:.1f}s")
                    except (ValueError, IndexError):
                        pass
        except Exception as e:
            log.debug(f"read_stdout: {e}")
    
    async def read_stderr():
        try:
            while True:
                chunk = await proc.stderr.read(4096)
                if not chunk: break
                stderr_buf.extend(chunk)
        except Exception as e:
            log.debug(f"read_stderr: {e}")
    
    # ⚠️ Niente communicate(): wait() non legge i pipe
    await asyncio.gather(read_stdout(), read_stderr(), proc.wait())
    
    if proc.returncode != 0:
        err = stderr_buf.decode(errors="ignore")[-500:]
        raise RuntimeError(f"FFmpeg failed ({proc.returncode}): {err}")


# ══════════════════════════════════════════════════════════════
# STEP 1 — EXTRACT SRT
# ══════════════════════════════════════════════════════════════
BASE_URL = "https://ca1.converter.app/mp4-to-srt"
SITE_URL = "https://converter.app"
HEADERS_WEB = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/130.0.0.0 Safari/537.36",
    "Referer": f"{SITE_URL}/", "Origin": SITE_URL,
}

def looks_like_srt(text):
    return bool(text and len(text.strip()) >= 10 and _TS_RE.search(text))

def extract_srt_from_html(html):
    """Estrae SRT da HTML con strategie multiple."""
    if not html: return None
    import html as _html
    
    # Strategia 1: cerca pattern SRT diretto in tutto l'HTML
    # (anche se contenuto in JSON, attributi, ecc)
    decoded = _html.unescape(html)
    
    # Cerca blocchi che iniziano con timestamp SRT
    srt_pattern = re.compile(
        r'(\d+\s*\n\s*\d{1,2}:\d{2}:\d{2}[.,]\d{3}\s*-->\s*'
        r'\d{1,2}:\d{2}:\d{2}[.,]\d{3}.*?)(?=(?:\n\s*\n\s*\d+\s*\n|\Z))',
        re.DOTALL
    )
    matches = srt_pattern.findall(decoded)
    if matches:
        # Ricostruisci unendo tutti i blocchi trovati
        joined = "\n\n".join(m.strip() for m in matches)
        if looks_like_srt(joined):
            return joined
    
    # Strategia 2: textarea/pre/code con decode entities
    for tag in ["textarea", "pre", "code"]:
        for m in re.finditer(
            rf'<{tag}[^>]*>(.*?)</{tag}>', html, re.S | re.I):
            content = _html.unescape(m.group(1)).strip()
            if looks_like_srt(content):
                return content
            # Prova anche a estrarre solo le righe SRT da dentro
            extracted = _extract_srt_lines(content)
            if extracted:
                return extracted
    
    # Strategia 3: cerca in <script> variabili JS
    for m in re.finditer(r'<script[^>]*>(.*?)</script>', html, re.S | re.I):
        script = m.group(1)
        # var srt = "..." o subtitle: "..."
        for sm in re.finditer(r'["\']([^"\']*\d{1,2}:\d{2}:\d{2}[.,]\d{3}\s*--&gt;[^"\']*)["\']', script):
            txt = _html.unescape(sm.group(1)).replace("\\n", "\n").replace("\\r", "")
            if looks_like_srt(txt):
                return txt
    
    # Strategia 4: cerca timestamp grezzi nel body e ricostruisci
    extracted = _extract_srt_lines(decoded)
    if extracted:
        return extracted
    
    return None


def _extract_srt_lines(text):
    """Cerca tutte le righe timestamp e i loro contenuti."""
    if not text: return None
    lines = text.split("\n")
    blocks = []
    current = []
    in_block = False
    
    ts_re = re.compile(r'^\s*\d{1,2}:\d{2}:\d{2}[.,]\d{3}\s*-->\s*\d{1,2}:\d{2}:\d{2}[.,]\d{3}')
    idx_re = re.compile(r'^\s*\d+\s*$')
    
    for line in lines:
        if ts_re.match(line):
            # Nuovo blocco
            if current and len(current) >= 2:
                blocks.append("\n".join(current))
            # cerca l'indice nelle righe precedenti (ultima riga numerica)
            prev_idx = str(len(blocks) + 1)
            current = [prev_idx, line.strip()]
            in_block = True
        elif in_block:
            stripped = line.strip()
            if not stripped:
                if len(current) >= 3:
                    blocks.append("\n".join(current))
                    current = []
                    in_block = False
            elif idx_re.match(stripped) and len(current) >= 3:
                # Inizio nuovo blocco senza riga vuota
                blocks.append("\n".join(current))
                current = []
                in_block = False
            else:
                # Pulisci tag HTML residui
                clean = re.sub(r'<[^>]+>', '', stripped).strip()
                if clean:
                    current.append(clean)
    
    if current and len(current) >= 3:
        blocks.append("\n".join(current))
    
    if not blocks: return None
    
    # Rinumera
    out = []
    for i, b in enumerate(blocks, 1):
        parts = b.split("\n")
        if len(parts) >= 2:
            out.append(f"{i}\n" + "\n".join(parts[1:]))
    
    result = "\n\n".join(out)
    return result if looks_like_srt(result) else None

def extract_url_from_html(html):
    for pat in [
        r'href=["\']([^"\']*\.srt[^"\']*)["\']',
        r'href=["\']([^"\']*download[^"\']*)["\']',
        r'(https?://[^\s"\'<>]*\.srt[^\s"\'<>]*)',
    ]:
        m = re.search(pat, html, re.I)
        if m: return m.group(1)
    return None

async def extract_audio_with_progress(input_path, jid, base_pct=0, span_pct=15):
    """
    Estrae solo audio mantenendo codec originale (-acodec copy).
    Output .mp4 container per compatibilità con converter.app.
    """
    out = os.path.join(TMP, f"dub_audio_{uuid.uuid4().hex}.mp4")
    duration = await get_video_duration_async(input_path)
    log.info(f"[{jid[:8]}] video duration: {duration:.1f}s")
    
    update(jid, step_progress=base_pct, message="Estrazione audio (stream copy)…")
    
    # Stream copy: veloce e mantiene il codec originale (no transcoding)
    cmd = [FFMPEG, "-y", "-i", input_path,
           "-vn", "-acodec", "copy", "-movflags", "+faststart",
           "-progress", "pipe:1", "-nostats", out]
    
    proc = await asyncio.create_subprocess_exec(
        *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
    
    stderr_buf = bytearray()
    
    async def read_stdout():
        try:
            while True:
                line = await proc.stdout.readline()
                if not line: break
                s = line.decode(errors="ignore").strip()
                if s.startswith("out_time_ms=") and duration > 0:
                    try:
                        us = int(s.split("=", 1)[1])
                        sec = us / 1_000_000
                        pct = base_pct + min(span_pct, (sec/duration) * span_pct)
                        actual_pct = sec/duration*100
                        update(jid, step_progress=round(pct, 2),
                               message=f"Audio extract {actual_pct:.2f}% ({sec:.2f}/{duration:.2f}s)")
                    except (ValueError, IndexError):
                        pass
        except Exception:
            pass
    
    async def read_stderr():
        try:
            while True:
                chunk = await proc.stderr.read(4096)
                if not chunk: break
                stderr_buf.extend(chunk)
        except Exception:
            pass
    
    await asyncio.gather(read_stdout(), read_stderr(), proc.wait())
    
    if proc.returncode != 0:
        # Fallback: prova con AAC encoding se stream copy fallisce
        log.warning(f"[{jid[:8]}] stream copy failed, trying AAC encode")
        err = stderr_buf.decode(errors="ignore")[-300:]
        log.warning(f"[{jid[:8]}] stream copy err: {err}")
        
        cmd2 = [FFMPEG, "-y", "-i", input_path, "-vn",
                "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", out]
        proc2 = await asyncio.create_subprocess_exec(
            *cmd2, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
        _, err2 = await proc2.communicate()
        if proc2.returncode != 0:
            raise RuntimeError(f"Audio extract failed: {err2.decode()[:300]}")
    
    size_mb = os.path.getsize(out) / 1048576
    log.info(f"[{jid[:8]}] audio extracted: {size_mb:.1f} MB")
    update(jid, step_progress=base_pct + span_pct,
           message=f"Audio pronto ({size_mb:.1f} MB)")
    return out

async def fetch_srt(session, jobid, server_filename=None, jid_log=""):
    """Versione semplificata + robusta basata sulla desktop funzionante."""
    out_name = None
    if server_filename:
        out_name = server_filename.rsplit(".", 1)[0] + ".srt"
    
    candidates = [
        f"{SITE_URL}/mp4-to-srt/result.php?jobid={jobid}&lang=en&w=ca1",
        f"{SITE_URL}/mp4-to-srt/result.php?jobid={jobid}",
        f"{BASE_URL}/result.php?jobid={jobid}&lang=en&w=ca1",
        f"{BASE_URL}/result.php?jobid={jobid}",
        f"{BASE_URL}/download.php?jobid={jobid}",
    ]
    if out_name:
        candidates.append(f"{BASE_URL}/download.php?jobid={jobid}&file={out_name}")
    
    req_timeout = aiohttp.ClientTimeout(total=10, sock_connect=5, sock_read=8)
    
    for idx, url in enumerate(candidates):
        try:
            log.info(f"[{jid_log}] fetch_srt try {idx+1}/{len(candidates)}: ...{url[-60:]}")
            async with session.get(url, allow_redirects=True, timeout=req_timeout) as resp:
                if resp.status not in (200, 206):
                    log.info(f"[{jid_log}]   → HTTP {resp.status}")
                    continue
                raw_bytes = await resp.read()
                text = raw_bytes.decode("utf-8", errors="ignore").strip()
                log.info(f"[{jid_log}]   → {len(text)} bytes, ctype={resp.headers.get('content-type','?')[:40]}")
                
                if looks_like_srt(text):
                    log.info(f"[{jid_log}]   ✅ SRT diretto!")
                    return text
                
                # Estrai da HTML (textarea/pre)
                emb = extract_srt_from_html(text)
                if emb:
                    log.info(f"[{jid_log}]   ✅ SRT estratto da HTML ({len(emb)} chars)")
                    return emb
                
                # Cerca link diretti
                dl = extract_url_from_html(text)
                if dl:
                    from urllib.parse import urljoin
                    dl = urljoin(str(resp.url), dl)
                    log.info(f"[{jid_log}]   → follow link: ...{dl[-60:]}")
                    try:
                        async with session.get(dl, allow_redirects=True, timeout=req_timeout) as dlr:
                            if dlr.status == 200:
                                t2 = (await dlr.read()).decode("utf-8", errors="ignore")
                                if looks_like_srt(t2):
                                    log.info(f"[{jid_log}]   ✅ SRT da link!")
                                    return t2
                    except Exception as e2:
                        log.info(f"[{jid_log}]   link fail: {type(e2).__name__}")
                else:
                    # Preview per debug
                    preview = re.sub(r'\s+', ' ', text[:300])
                    log.info(f"[{jid_log}]   preview: {preview}")
        except asyncio.TimeoutError:
            log.info(f"[{jid_log}]   ⏱ timeout")
            continue
        except Exception as e:
            log.info(f"[{jid_log}]   ✗ {type(e).__name__}: {str(e)[:80]}")
            continue
    
    log.warning(f"[{jid_log}] fetch_srt: nessun candidato ha funzionato")
    return None

async def step1_extract_srt(video_path, jid):
    update(jid, step=1, stage="extract", step_progress=0, message="Inizio…")
    audio_path = await extract_audio_with_progress(video_path, jid, 0, 15)
    
    try:
        file_size = os.path.getsize(audio_path)
        # IMPORTANTE: usa il filename del VIDEO ORIGINALE, non dell'audio temp
        # converter.app riconosce meglio i nomi tipo "movie.mp4"
        filename = os.path.basename(video_path)
        if not filename.lower().endswith(('.mp4', '.mkv', '.avi', '.mov', '.webm')):
            filename = filename + ".mp4"
        
        log.info(f"[{jid[:8]}] uploading to converter.app: {file_size/1048576:.1f} MB as '{filename}'")
        
        # Boundary stile WebKit come fa il browser
        boundary = f"----WebKitFormBoundary{os.urandom(8).hex()}"
        pre = (f"--{boundary}\r\n"
               f'Content-Disposition: form-data; name="files[]"; filename="{filename}"\r\n'
               f"Content-Type: video/mp4\r\n\r\n").encode()
        post = (f"\r\n--{boundary}\r\n"
                f'Content-Disposition: form-data; name="ajax"\r\n\r\n'
                f"1\r\n--{boundary}--\r\n").encode()
        total = len(pre) + file_size + len(post)

        async def sender():
            yield pre
            async with aiofiles.open(audio_path, "rb") as f:
                sent = len(pre)
                last_t = 0
                while True:
                    chunk = await f.read(CHUNK_SIZE)
                    if not chunk: break
                    sent += len(chunk)
                    now = time.time()
                    if now - last_t > 0.1:
                        pct = 15 + (sent/total) * 25
                        update(jid, step_progress=round(pct, 2),
                               message=f"Upload audio {sent/1048576:.1f}/{total/1048576:.1f} MB")
                        last_t = now
                    yield chunk
            yield post

        headers = {**HEADERS_WEB,
                   "Content-Type": f"multipart/form-data; boundary={boundary}",
                   "Content-Length": str(total)}
        
        upload_timeout = aiohttp.ClientTimeout(total=600, sock_connect=20, sock_read=120)
        poll_timeout = aiohttp.ClientTimeout(total=20, sock_connect=10, sock_read=15)
        connector = aiohttp.TCPConnector(limit=10, ttl_dns_cache=300)

        async with aiohttp.ClientSession(connector=connector, headers=HEADERS_WEB) as session:
            log.info(f"[{jid[:8]}] POST uploader.php")
            try:
                async with session.post(f"{BASE_URL}/uploader.php",
                                        data=sender(), headers=headers,
                                        timeout=upload_timeout) as resp:
                    if resp.status != 200:
                        raise Exception(f"Upload HTTP {resp.status}")
                    upload_text = await resp.text()
                    log.info(f"[{jid[:8]}] upload response: {upload_text[:300]}")
            except asyncio.TimeoutError:
                raise Exception("Timeout upload converter.app")

            update(jid, step_progress=42, message="Server in elaborazione…")

            jobid = server_filename = None
            try:
                data = json.loads(upload_text)
                jobid = data.get("jobid")
                server_filename = data.get("filename")
                if not jobid and data.get("files"):
                    fi = data["files"][0]
                    jobid = fi.get("jobid")
                    server_filename = fi.get("name", fi.get("filename"))
                    if not jobid and "url" in fi:
                        m = re.search(r"jobid=([a-zA-Z0-9]+)", fi["url"])
                        jobid = m.group(1) if m else None
            except: pass
            if not jobid:
                m = re.search(r'jobid[=:"\s]+([a-zA-Z0-9]+)', upload_text)
                jobid = m.group(1) if m else None
            if not jobid:
                raise Exception(f"No jobid received: {upload_text[:200]}")

            log.info(f"[{jid[:8]}] remote jobid: {jobid}, server filename: {server_filename}")
            update(jid, message=f"Job remoto {jobid[:8]}…")
            
            try:
                async with session.post(f"{BASE_URL}/process.php?jobid={jobid}",
                                        timeout=poll_timeout) as r:
                    await r.text()
            except: pass

            STATUS = {0:5, 1:10, 2:40, 3:60, 4:80, 5:100, 6:100,
                      100:100, 101:100, 200:100, 99:-1}
            ts = int(time.time()*1000)
            poll = 1.0
            last_pct = -1
            last_change_t = time.time()
            poll_errors = 0
            
            for i in range(1500):
                if is_cancelled(jid): raise InterruptedError("cancelled")
                await asyncio.sleep(poll)
                
                if i % 5 == 0:
                    log.info(f"[{jid[:8]}] poll #{i} (last_pct={last_pct}, errors={poll_errors})")
                
                raw = None
                try:
                    async with session.get(
                        f"{BASE_URL}/display.php?jobid={jobid}&_={ts+i}",
                        timeout=poll_timeout) as r:
                        raw = (await r.text()).strip()
                    poll_errors = 0
                except asyncio.TimeoutError:
                    poll_errors += 1
                    log.warning(f"[{jid[:8]}] poll timeout #{poll_errors}")
                    if poll_errors >= 5 and last_pct >= 60:
                        break
                    continue
                except Exception as e:
                    poll_errors += 1
                    if poll_errors >= 10: 
                        raise Exception(f"Troppi errori polling: {e}")
                    continue
                
                try: code = int(json.loads(raw))
                except: 
                    log.warning(f"[{jid[:8]}] bad poll response: {raw[:100]}")
                    continue
                
                if code in STATUS:
                    pct = STATUS[code]
                elif code >= 5:
                    pct = 100
                elif code < 0:
                    pct = -1
                else:
                    pct = min(max(code * 20, 5), 90)
                
                if pct == -1: raise Exception(f"Server code={code}")
                
                step_pct = 45 + (pct/100) * 50
                update(jid, step_progress=round(step_pct, 2),
                       message=f"Trascrizione: {pct}% (poll #{i})")
                
                if pct != last_pct:
                    log.info(f"[{jid[:8]}] transcribe {pct}% (code={code})")
                    last_pct = pct
                    last_change_t = time.time()
                else:
                    stuck = time.time() - last_change_t
                    if stuck > 120 and last_pct >= 60:
                        log.warning(f"[{jid[:8]}] stuck at {last_pct}% for {stuck:.0f}s, forcing fetch")
                        break
                
                poll = 0.6 if pct < 80 else (1.0 if pct < 100 else 0.5)
                
                if pct >= 100:
                    log.info(f"[{jid[:8]}] reached 100%, fetching SRT")
                    await asyncio.sleep(0.5)
                    break
            
            update(jid, step_progress=98, message="Recupero SRT…")
            log.info(f"[{jid[:8]}] starting SRT recovery loop (30 attempts)")
            srt = None
            for attempt in range(30):
                if is_cancelled(jid): raise InterruptedError("cancelled")
                log.info(f"[{jid[:8]}] === SRT recovery attempt {attempt+1}/30 ===")
                try:
                    srt = await asyncio.wait_for(
                        fetch_srt(session, jobid, server_filename, jid[:8]),
                        timeout=60)
                    if srt: break
                except asyncio.TimeoutError:
                    log.warning(f"[{jid[:8]}] fetch_srt timeout attempt {attempt+1}")
                except Exception as e:
                    log.warning(f"[{jid[:8]}] fetch_srt err: {e}")
                update(jid, message=f"Recupero SRT tentativo {attempt+1}/30…")
                await asyncio.sleep(1.5)
            
            if not srt:
                raise Exception("SRT non recuperabile dopo 30 tentativi")
            
            log.info(f"[{jid[:8]}] SRT ok: {len(srt)} chars")
            update(jid, step_progress=100, message=f"SRT estratto ({len(srt)} char)")
            return srt
    finally:
        if os.path.exists(audio_path):
            try: os.remove(audio_path)
            except: pass


# ══════════════════════════════════════════════════════════════
# STEP 2 — TRANSLATE
# ══════════════════════════════════════════════════════════════
class TransCache:
    __slots__ = ("_c", "_max")
    def __init__(self, max_size=2000):
        self._c = OrderedDict(); self._max = max_size
    def get(self, t, tg):
        k = hashlib.md5(f"{tg}|{t}".encode()).digest()
        v = self._c.get(k)
        if v is not None: self._c.move_to_end(k)
        return v
    def put(self, t, tg, r):
        if r.strip() == t.strip(): return
        k = hashlib.md5(f"{tg}|{t}".encode()).digest()
        self._c[k] = r
        while len(self._c) > self._max: self._c.popitem(last=False)

_CACHE = TransCache()

async def _t_google(client, text, tgt):
    r = await client.get("https://translate.googleapis.com/translate_a/single",
        params={"client":"gtx","sl":"auto","tl":tgt,"dt":"t","q":text}, timeout=15)
    r.raise_for_status()
    data = r.json()
    out = ""
    if isinstance(data, list) and data[0]:
        for c in data[0]:
            if c and c[0]: out += c[0]
    if out: return out
    raise RuntimeError("empty")

async def _t_mymemory(client, text, tgt):
    r = await client.get("https://api.mymemory.translated.net/get",
        params={"q": text[:500], "langpair": f"auto|{tgt}"}, timeout=15)
    r.raise_for_status()
    d = r.json()
    if d.get("responseStatus") == 200:
        out = d["responseData"]["translatedText"]
        if out and "MYMEMORY WARNING" not in out: return out
    raise RuntimeError("fail")

async def _t_libre(client, text, tgt):
    for base in ["https://libretranslate.com", "https://translate.argosopentech.com"]:
        try:
            r = await client.post(f"{base}/translate",
                json={"q": text[:1000], "source":"auto", "target":tgt, "format":"text"}, timeout=10)
            if r.status_code == 200:
                out = r.json().get("translatedText", "")
                if out: return out
        except: continue
    raise RuntimeError("fail")

ASYNC_APIS = [_t_google, _t_mymemory, _t_libre]

async def translate_one(client, text, tgt):
    if not text or not text.strip(): return text
    text = text.strip()
    c = _CACHE.get(text, tgt)
    if c: return c
    for fn in ASYNC_APIS:
        try:
            r = (await fn(client, text, tgt)).strip()
            if not r: continue
            if re.sub(r'\s+', ' ', text.lower()) == re.sub(r'\s+', ' ', r.lower()): continue
            _CACHE.put(text, tgt, r)
            return r
        except: continue
    return text

async def step2_translate(srt_content, target_lang, jid):
    update(jid, step=2, stage="translate", step_progress=0, message=f"Traduzione → {target_lang}")
    blocks = parse_srt(srt_content)
    if not blocks: raise Exception("SRT vuoto")
    total = len(blocks)
    log.info(f"[{jid[:8]}] translating {total} blocks → {target_lang}")
    done = [0]
    sem = asyncio.Semaphore(TRANS_CONCURRENCY)
    lock = asyncio.Lock()

    async with httpx.AsyncClient(
        headers={"User-Agent":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/130.0.0.0 Safari/537.36"},
        limits=httpx.Limits(max_connections=TRANS_CONCURRENCY*3, max_keepalive_connections=TRANS_CONCURRENCY*2),
        timeout=httpx.Timeout(15.0, connect=5.0),
        http2=False,  # gtx non supporta HTTP/2 bene
    ) as client:
        async def worker(b):
            async with sem:
                if is_cancelled(jid): raise InterruptedError("cancelled")
                b["text"] = await translate_one(client, b["text"], target_lang)
                async with lock:
                    done[0] += 1
                    d = done[0]
                    pct = d / total * 100
                update(jid, step_progress=round(pct, 2),
                       message=f"Traduzione {d}/{total} ({pct:.1f}%)")
        await asyncio.gather(*(worker(b) for b in blocks))
    log.info(f"[{jid[:8]}] translation done ({total} blocks)")
    return build_srt(blocks)


# ══════════════════════════════════════════════════════════════
# STEP 3 — TTS
# ══════════════════════════════════════════════════════════════
def get_voice(lang, gender):
    key = (lang.lower(), gender.capitalize())
    if key in VOICES: return VOICES[key]
    for (l, _), v in VOICES.items():
        if l == lang.lower(): return v
    return "en-US-AriaNeural"

def get_duration(path):
    try:
        r = subprocess.run([FFPROBE, "-v", "error", "-show_entries",
                            "format=duration", "-of",
                            "default=noprint_wrappers=1:nokey=1", path],
                           capture_output=True, text=True, timeout=10)
        return float(r.stdout.strip() or 0)
    except: return 0.0

def gen_silence(dur, out):
    if dur <= 0: dur = 0.01
    subprocess.run([FFMPEG, "-y", "-f", "lavfi",
                    "-i", "anullsrc=r=24000:cl=mono", "-t", f"{dur:.4f}",
                    "-c:a", "libmp3lame", "-b:a", "48k", "-ar", "24000", "-ac", "1", out],
                   capture_output=True, check=True)

def speed_fit(inp, out, target):
    actual = get_duration(inp)
    if actual <= 0:
        shutil.copy2(inp, out); return
    ratio = actual / target
    if ratio < 0.5: ratio = 0.5
    if ratio > 100: ratio = 100
    filters = []
    r = ratio
    while r > 2.0: filters.append("atempo=2.0"); r /= 2.0
    while r < 0.5: filters.append("atempo=0.5"); r /= 0.5
    filters.append(f"atempo={r:.6f}")
    res = subprocess.run([FFMPEG, "-y", "-i", inp, "-filter:a", ",".join(filters),
           "-c:a", "libmp3lame", "-b:a", "64k", "-ar", "24000", "-ac", "1",
           "-threads", "1", out], capture_output=True)
    if res.returncode != 0:
        shutil.copy2(inp, out)

_TTS_CACHE_DIR = os.path.join(TMP, "dubflow_tts_cache")
os.makedirs(_TTS_CACHE_DIR, exist_ok=True)


def _tts_cache_key(text, voice, rate, pitch, volume):
    """Genera chiave cache stabile per il TTS."""
    key = f"{voice}|{rate}|{pitch}|{volume}|{text}"
    return hashlib.md5(key.encode()).hexdigest()


# Import gTTS UNA VOLTA al boot (lazy)
_GTTS_CLASS = None

def _get_gtts():
    global _GTTS_CLASS
    if _GTTS_CLASS is None:
        try:
            from gtts import gTTS as _g
            _GTTS_CLASS = _g
        except ImportError:
            import subprocess as sp
            import sys as _sys
            sp.check_call([_sys.executable, "-m", "pip", "install", "-q", "gTTS"])
            from gtts import gTTS as _g
            _GTTS_CLASS = _g
    return _GTTS_CLASS


_GTTS_LANG_MAP = {"nb": "no", "zh": "zh-cn"}


async def _gtts_fallback(text, lang_code, out):
    """Fallback gTTS ottimizzato: import cached + hard link cache."""
    loop = asyncio.get_event_loop()
    gTTS = _get_gtts()
    lang = _GTTS_LANG_MAP.get(lang_code, lang_code)
    
    def _sync_gtts():
        try:
            tts = gTTS(text=text, lang=lang, slow=False, tld="com")
            tts.save(out)
        except Exception:
            tts = gTTS(text=text, lang="en", slow=False)
            tts.save(out)
    
    await loop.run_in_executor(THREAD_POOL, _sync_gtts)
    
    try:
        if os.stat(out).st_size < 100:
            raise RuntimeError("gTTS empty file")
    except FileNotFoundError:
        raise RuntimeError("gTTS no file produced")


# Flag globale: una volta che edge è bloccato, non riprovare
_EDGE_BLOCKED = False


async def tts_block(text, voice, out, rate, pitch, volume):
    """TTS ottimizzato: cache, edge-tts con skip rapido, gTTS fallback."""
    global _EDGE_BLOCKED
    
    # 1. Check cache (1 syscall invece di 2)
    cache_key = _tts_cache_key(text, voice, rate, pitch, volume)
    cache_path = os.path.join(_TTS_CACHE_DIR, f"{cache_key}.mp3")
    try:
        st = os.stat(cache_path)
        if st.st_size > 100:
            # Hard link invece di copy quando possibile (instant)
            try:
                os.link(cache_path, out)
            except (OSError, AttributeError):
                shutil.copy2(cache_path, out)
            return
    except (FileNotFoundError, OSError):
        pass
    
    lang_code = voice.split("-")[0].lower() if "-" in voice else "en"
    
    # 2. Edge-tts (skip se già confermato bloccato)
    if not _EDGE_BLOCKED:
        try:
            comm = edge_tts.Communicate(
                text=text, voice=voice,
                rate=rate, pitch=pitch, volume=volume,
            )
            data = bytearray()
            async for c in comm.stream():
                if c["type"] == "audio":
                    data.extend(c["data"])
            if data:
                async with aiofiles.open(out, "wb") as f:
                    await f.write(bytes(data))
                # Salva in cache (hard link veloce)
                try:
                    if not os.path.exists(cache_path):
                        os.link(out, cache_path)
                except (OSError, AttributeError):
                    try: shutil.copy2(out, cache_path)
                    except: pass
                return
        except Exception as e:
            if "403" in str(e) or "Invalid response" in str(e):
                _EDGE_BLOCKED = True
                log.warning("edge-tts permanently blocked on this IP, using gTTS only")
    
    # 3. Fallback gTTS
    await _gtts_fallback(text, lang_code, out)
    # Cache anche il risultato gTTS (hard link veloce)
    try:
        if not os.path.exists(cache_path):
            os.link(out, cache_path)
    except (OSError, AttributeError):
        try: shutil.copy2(out, cache_path)
        except: pass


async def step3_tts(srt, lang, gender, out_mp3, jid, rate, pitch, volume):
    update(jid, step=3, stage="tts", step_progress=0, message="Avvio TTS…")
    blocks = parse_srt(srt)
    if not blocks: raise Exception("no blocks")
    voice = get_voice(lang, gender)
    log.info(f"[{jid[:8]}] TTS start: {len(blocks)} blocks, voice={voice}")
    tmp = os.path.join(TMP, f"dub_tts_{uuid.uuid4().hex}")
    os.makedirs(tmp, exist_ok=True)
    sem = asyncio.Semaphore(TTS_CONCURRENCY)
    total = len(blocks)
    done = [0]
    lock = asyncio.Lock()
    loop = asyncio.get_event_loop()
    results = [None] * total
    t_start = time.time()

    async def process(i, b):
        async with sem:
            if is_cancelled(jid): raise InterruptedError("cancelled")
            text = clean_text(b["text"])
            slot = b["end_sec"] - b["start_sec"]
            if not text or slot <= 0:
                async with lock:
                    done[0] += 1
                    d = done[0]
                pct = d / total * 85
                update(jid, step_progress=round(pct, 2),
                       message=f"TTS {d}/{total} (skip vuoto)")
                return
            raw = os.path.join(tmp, f"b{i:04d}_raw.mp3")
            fit = os.path.join(tmp, f"b{i:04d}_fit.mp3")
            try:
                await tts_block(text, voice, raw, rate, pitch, volume)
            except Exception as e:
                log.warning(f"[{jid[:8]}] TTS block {i}: {e}")
                async with lock:
                    done[0] += 1
                    d = done[0]
                pct = d / total * 85
                update(jid, step_progress=round(pct, 2),
                       message=f"TTS {d}/{total} (errore blocco)")
                return
            actual = await loop.run_in_executor(THREAD_POOL, get_duration, raw)
            if actual > slot * 1.05:
                await loop.run_in_executor(THREAD_POOL, speed_fit, raw, fit, slot)
            else:
                await loop.run_in_executor(THREAD_POOL, shutil.copy2, raw, fit)
            results[i] = (fit, b["start_sec"], slot)
            
            async with lock:
                done[0] += 1
                d = done[0]
            pct = d / total * 85
            elapsed = time.time() - t_start
            rate_per_sec = d / elapsed if elapsed > 0 else 0
            eta = (total - d) / rate_per_sec if rate_per_sec > 0 else 0
            update(jid, step_progress=round(pct, 2),
                   message=f"TTS {d}/{total} · {rate_per_sec:.1f} blk/s · ETA {eta:.0f}s")
            
            # Log ogni 5 blocchi per visibilità
            if d % 5 == 0 or d == total:
                log.info(f"[{jid[:8]}] TTS {d}/{total} ({pct:.1f}%) · {rate_per_sec:.1f} blk/s · ETA {eta:.0f}s")

    try:
        await asyncio.gather(*(process(i, b) for i, b in enumerate(blocks)))
        segs = [r for r in results if r]
        if not segs: raise Exception("no segments")
        update(jid, step_progress=87, message=f"Assemblaggio {len(segs)} segmenti…")
        log.info(f"[{jid[:8]}] TTS done in {time.time()-t_start:.1f}s, assembling {len(segs)} segments")
        await loop.run_in_executor(THREAD_POOL, _assemble, segs, blocks, tmp, out_mp3)
        update(jid, step_progress=100, message="Audio sincronizzato pronto")
        log.info(f"[{jid[:8]}] TTS+assembly total: {time.time()-t_start:.1f}s")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

def _assemble(segs, blocks, tmp, out_mp3):
    last_end = max(b["end_sec"] for b in blocks)
    segs_s = sorted(segs, key=lambda x: x[1])
    padded = os.path.join(tmp, "padded"); os.makedirs(padded, exist_ok=True)
    concat = []
    current = 0.0
    for idx, (seg, start, slot) in enumerate(segs_s):
        gap = start - current
        if gap > 0.01:
            sp = os.path.join(padded, f"sil_{idx:04d}.mp3")
            gen_silence(gap, sp); concat.append(sp); current += gap
        concat.append(seg)
        sd = get_duration(seg) or slot
        current += sd
    rem = last_end - current
    if rem > 0.1:
        tp = os.path.join(padded, "sil_trail.mp3")
        gen_silence(rem, tp); concat.append(tp)
    listf = os.path.join(tmp, "concat.txt")
    with open(listf, "w") as f:
        for p in concat:
            safe = p.replace("\\", "/").replace("'", "'\\''")
            f.write(f"file '{safe}'\n")
    res = subprocess.run([FFMPEG, "-y", "-f", "concat", "-safe", "0", "-i", listf,
           "-c:a", "libmp3lame", "-b:a", "128k", "-ar", "24000", "-ac", "1",
           "-threads", str(FFMPEG_THREADS), out_mp3], capture_output=True)
    if res.returncode != 0:
        raise RuntimeError(f"concat: {res.stderr.decode()[:300]}")


# ══════════════════════════════════════════════════════════════
# STEP 4 — DUB (FIXED pipe handling)
# ══════════════════════════════════════════════════════════════
async def step4_dub(video, audio, out, mode, dub_vol, orig_vol, fade, jid):
    update(jid, step=4, stage="dub", step_progress=5, message=f"Dub ({mode})")
    duration = await get_video_duration_async(video)
    log.info(f"[{jid[:8]}] dub duration={duration:.1f}s mode={mode}")
    
    base = [FFMPEG, "-y", "-i", video, "-i", audio, "-c:v", "copy",
            "-threads", str(FFMPEG_THREADS), "-movflags", "+faststart",
            "-progress", "pipe:1", "-nostats"]
    if mode == "replace":
        cmd = base + ["-map", "0:v:0", "-map", "1:a:0",
                      "-c:a", "aac", "-b:a", "192k", "-shortest", out]
    elif mode == "mix":
        fc = (f"[0:a]volume={orig_vol}[o];[1:a]volume={dub_vol}[d];"
              f"[o][d]amix=inputs=2:duration=shortest:dropout_transition=2[a]")
        cmd = base + ["-filter_complex", fc, "-map", "0:v:0", "-map", "[a]",
                      "-c:a", "aac", "-b:a", "192k", out]
    else:
        mn = min(duration, get_duration(audio)) or 300
        fo = max(0, mn - fade)
        fc = (f"[0:a]volume={orig_vol}[o];"
              f"[1:a]volume={dub_vol},afade=t=in:st=0:d={fade},"
              f"afade=t=out:st={fo}:d={fade}[d];"
              f"[o][d]amix=inputs=2:duration=shortest:dropout_transition=2,"
              f"loudnorm=I=-16:TP=-1.5:LRA=11[a]")
        cmd = base + ["-filter_complex", fc, "-map", "0:v:0", "-map", "[a]",
                      "-c:a", "aac", "-b:a", "192k", out]

    await run_ffmpeg_with_progress(cmd, duration, jid, 10, 85, "FFmpeg ")
    update(jid, step_progress=100, message="Dub completato")
    log.info(f"[{jid[:8]}] dub completed")


# ══════════════════════════════════════════════════════════════
# PIPELINE
# ══════════════════════════════════════════════════════════════
# ══════════════════════════════════════════════════════════════
# SSE PROGRESS
# ══════════════════════════════════════════════════════════════

async def run_pipeline(jid, base_url):
    j = JOBS[jid]
    log.info(f"[{jid[:8]}] pipeline waiting for video…")
    
    wait_start = time.time()
    while not j["video_ready"]:
        if j.get("error") or j["cancel"]: 
            log.info(f"[{jid[:8]}] pipeline aborted before video")
            return
        if time.time() - wait_start > 1200:
            update(jid, error="Timeout attesa video", done=True)
            return
        await asyncio.sleep(0.01)
    
    log.info(f"[{jid[:8]}] video ready, starting pipeline")
    # Forza step 0 = done al 100% prima di iniziare step 1
    update(jid, step=0, step_progress=100, message="Video ricevuto, avvio pipeline…")
    await asyncio.sleep(0.1)  # piccolo delay per far passare l'update SSE
    video_path = j["video_path"]
    p = j["params"]
    target_lang = p["lang"]
    
    # NOMI INTELLIGENTI in italiano basati su nome originale
    original = j.get("original_name", "video")
    safe_original = re.sub(r'[^\w\-_.]', '_', original)[:60] or "video"
    # Rimuovi underscore multipli e di troppo
    safe_original = re.sub(r'_+', '_', safe_original).strip('_') or "video"
    uid_short = uuid.uuid4().hex[:6]
    
    out_mp4_name = f"{safe_original}_doppiato_{target_lang}_{uid_short}.mp4"
    srt_orig_name = f"{safe_original}_sottotitoli_{uid_short}.srt"
    srt_trans_name = f"{safe_original}_traduzione_{target_lang}_{uid_short}.srt"
    
    out_path = os.path.join(PUBLIC, out_mp4_name)
    srt_path = os.path.join(PUBLIC, srt_orig_name)
    tsrt_path = os.path.join(PUBLIC, srt_trans_name)
    tts_path = os.path.join(TMP, f"dub_tts_{uid_short}.mp3")

    t0 = time.time()
    try:
        srt = await step1_extract_srt(video_path, jid)
        async with aiofiles.open(srt_path, "w", encoding="utf-8") as f:
            await f.write(srt)

        tsrt = await step2_translate(srt, target_lang, jid)
        async with aiofiles.open(tsrt_path, "w", encoding="utf-8") as f:
            await f.write(tsrt)

        await step3_tts(tsrt, target_lang, p["gender"], tts_path, jid,
                        p["rate"], p["pitch"], p["volume"])

        await step4_dub(video_path, tts_path, out_path,
                        p["mode"], p["dub_volume"], p["orig_volume"], p["fade"], jid)
        
        elapsed = time.time() - t0
        update(jid, stage="done", done=True, step=4, step_progress=100,
               message=f"Completato in {elapsed:.2f}s",
               result={
                   "video_url": f"{base_url}/file/{out_mp4_name}",
                   "video_download": f"{base_url}/file/{out_mp4_name}?dl=1",
                   "video_filename": out_mp4_name,
                   "srt_original": f"{base_url}/file/{srt_orig_name}",
                   "srt_original_filename": srt_orig_name,
                   "srt_translated": f"{base_url}/file/{srt_trans_name}",
                   "srt_filename": srt_trans_name,
                   "language": target_lang, "gender": p["gender"],
                   "elapsed_sec": round(elapsed, 2),
               })
        JOBS[jid]["progress"] = 100.0
        log.info(f"[{jid[:8]}] ✅ done in {elapsed:.2f}s → {out_mp4_name}")
    except InterruptedError:
        log.info(f"[{jid[:8]}] cancelled")
        update(jid, error="Cancellato", done=True)
    except Exception as e:
        log.exception(f"[{jid[:8]}] ❌ failed: {e}")
        update(jid, error=str(e)[:300], done=True)
    finally:
        cleanup_paths([video_path, tts_path])



@app.get("/progress/{jid}", tags=["Progress"])
async def progress(jid: str):
    """Server-Sent Events stream con stato job in real-time."""
    async def gen():
        last_send = None
        last_time = time.time()
        while True:
            j = JOBS.get(jid)
            if not j:
                yield f"data: {json.dumps({'error':'job not found'})}\n\n"
                return
            payload = {k: j[k] for k in
                       ("progress","step","step_progress","message","stage","done","error","result")}
            s = json.dumps(payload)
            now = time.time()
            # invia sempre se cambia, max 20Hz; heartbeat ogni 1.5s
            if s != last_send:
                yield f"data: {s}\n\n"
                last_send = s
                last_time = now
            elif (now - last_time) > 1.5:
                yield f"data: {s}\n\n"
                last_time = now
            if j["done"] or j["error"]: 
                return
            await asyncio.sleep(0.02)
    return StreamingResponse(gen(), media_type="text/event-stream",
                             headers={"Cache-Control":"no-cache",
                                      "X-Accel-Buffering":"no",
                                      "Connection":"keep-alive"})


@app.get("/status/{jid}", tags=["Progress"])
async def status(jid: str):
    """Polling endpoint (per chi non usa SSE)."""
    j = JOBS.get(jid)
    if not j: raise HTTPException(404, "job not found")
    return {k: j[k] for k in
            ("progress","step","step_progress","message","stage","done","error","result","created")}


@app.post("/cancel/{jid}", tags=["Jobs"])
async def cancel(jid: str):
    if jid in JOBS:
        JOBS[jid]["cancel"] = True
        log.info(f"[{jid[:8]}] cancel requested")
        return {"ok": True}
    raise HTTPException(404, "not found")


# ══════════════════════════════════════════════════════════════
# JOB ENDPOINTS — Two-step (create + upload/url)
# ══════════════════════════════════════════════════════════════
def make_params(lang, gender, mode, dub_volume, orig_volume, fade, rate, pitch, volume):
    return {
        "lang": lang, "gender": gender, "mode": mode,
        "dub_volume": float(dub_volume), "orig_volume": float(orig_volume), "fade": float(fade),
        "rate": rate, "pitch": pitch, "volume": volume,
    }


@app.post("/job/create", tags=["Jobs"])
async def job_create(request: Request,
                     lang: str = "it", gender: str = "Female", mode: str = "replace",
                     dub_volume: float = 0.85, orig_volume: float = 0.15, fade: float = 1.0,
                     rate: str = "+0%", pitch: str = "+0Hz", volume: str = "+0%"):
    """Crea job vuoto. Poi chiama /job/{jid}/upload o /job/{jid}/url."""
    jid = new_job()
    JOBS[jid]["params"] = make_params(lang, gender, mode, dub_volume, orig_volume, fade, rate, pitch, volume)
    update(jid, stage="upload", step=0, step_progress=0, message="In attesa video…")
    base = str(request.base_url).rstrip("/")
    asyncio.create_task(run_pipeline(jid, base))
    return {"job_id": jid, "status_url": f"{base}/status/{jid}", "progress_url": f"{base}/progress/{jid}"}
@app.post("/job/{jid}/upload", tags=["Jobs"])
async def job_upload(jid: str, video: UploadFile = File(...)):
    j = JOBS.get(jid)
    if not j: raise HTTPException(404, "job not found")
    if j["video_ready"]: raise HTTPException(400, "video già caricato")
    
    # Estrai nome originale (senza path/estensione cattiva)
    original_name = os.path.basename(video.filename or "video.mp4")
    name_stem = os.path.splitext(original_name)[0]
    # Sanitize: solo caratteri sicuri per filesystem
    name_stem = re.sub(r'[^\w\-_.]', '_', name_stem)[:80] or "video"
    j["original_name"] = name_stem
    log.info(f"[{jid[:8]}] original filename: '{original_name}' → stem: '{name_stem}'")
    
    uid = uuid.uuid4().hex
    vpath = os.path.join(TMP, f"dub_in_{uid}.mp4")
    size = 0
    
    try:
        async with aiofiles.open(vpath, "wb") as f:
            last_t = 0
            while True:
                c = await video.read(CHUNK_SIZE * 4)
                if not c: break
                await f.write(c)
                size += len(c)
                if size > MAX_UPLOAD_MB * 1024 * 1024:
                    raise HTTPException(413, f"Max {MAX_UPLOAD_MB} MB")
                now = time.time()
                if now - last_t > 0.05:
                    update(jid, step=0, step_progress=min(95, size/(MAX_UPLOAD_MB*1024*1024)*95),
                           message=f"Ricezione server {size/1048576:.2f} MB")
                    last_t = now
        
        if size < MIN_FILE_BYTES:
            raise HTTPException(400, "File troppo piccolo")
        
        update(jid, step_progress=97, message="Validazione…")
        await validate_video(vpath)
        log.info(f"[{jid[:8]}] upload OK: {size/1048576:.2f} MB")
        update(jid, step=0, step_progress=100, message=f"Video pronto ({size/1048576:.2f} MB)",
               video_path=vpath, video_ready=True)
        return {"ok": True, "size_mb": round(size/1048576, 2), "name": name_stem}
    except HTTPException as e:
        cleanup_paths([vpath])
        update(jid, error=str(e.detail), done=True)
        raise
    except Exception as e:
        log.exception(f"[{jid[:8]}] upload failed")
        cleanup_paths([vpath])
        update(jid, error=str(e), done=True)
        raise HTTPException(500, str(e))
@app.post("/job/{jid}/url", tags=["Jobs"])
async def job_url(jid: str, video: str = Query(...)):
    j = JOBS.get(jid)
    if not j: raise HTTPException(404, "job not found")
    if j["video_ready"]: raise HTTPException(400, "video già caricato")
    
    video = video.strip()
    if video.startswith("blob:"):
        update(jid, error="blob: non supportato", done=True)
        raise HTTPException(400, "blob: non scaricabile. Usa Upload.")
    if not re.match(r'^https?://', video, re.I):
        update(jid, error="URL non valido", done=True)
        raise HTTPException(400, "URL deve iniziare con http/https")
    
    from urllib.parse import urlparse, unquote
    parsed = urlparse(video)
    url_path = unquote(parsed.path)
    original_name = os.path.basename(url_path) or "video.mp4"
    
    # Se è un UUID o nome generico senza estensione, prova HEAD per content-disposition
    if "." not in original_name or len(original_name) < 5:
        name_stem = "video"
    else:
        name_stem = os.path.splitext(original_name)[0]
    name_stem = re.sub(r'[^\w\-_.]', '_', name_stem)[:80] or "video"
    j["original_name"] = name_stem
    log.info(f"[{jid[:8]}] URL: {video}")
    log.info(f"[{jid[:8]}] URL filename: '{original_name}' → stem: '{name_stem}'")
    
    uid = uuid.uuid4().hex
    vpath = os.path.join(TMP, f"dub_in_{uid}.mp4")
    
    # Headers browser-like per HF Spaces, CDN, etc
    browser_headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
                      "(KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36",
        "Accept": "video/mp4,video/*,*/*;q=0.8",
        "Accept-Language": "en-US,en;q=0.9,it;q=0.8",
        "Accept-Encoding": "identity",  # niente compressione per video binari
        "Connection": "keep-alive",
        "Sec-Fetch-Dest": "video",
        "Sec-Fetch-Mode": "no-cors",
        "Sec-Fetch-Site": "cross-site",
    }
    
    try:
        update(jid, step=0, step_progress=2, message="Connessione…")
        
        # Retry per HF Spaces in cold start
        last_error = None
        for attempt in range(3):
            try:
                async with httpx.AsyncClient(
                    timeout=httpx.Timeout(60.0, connect=30.0, read=600.0),
                    follow_redirects=True,
                    headers=browser_headers,
                ) as c:
                    # Prima HEAD per dimensione + content-disposition (filename vero)
                    try:
                        head_resp = await c.head(video, timeout=15.0)
                        if "content-disposition" in head_resp.headers:
                            cd = head_resp.headers["content-disposition"]
                            m = re.search(r'filename[*]?=["\']?([^"\';]+)', cd)
                            if m:
                                real_name = m.group(1).strip().strip('"').strip("'")
                                if real_name.startswith("UTF-8''"):
                                    real_name = unquote(real_name[7:])
                                stem = os.path.splitext(os.path.basename(real_name))[0]
                                stem = re.sub(r'[^\w\-_.]', '_', stem)[:80]
                                if stem and stem != "video":
                                    j["original_name"] = stem
                                    log.info(f"[{jid[:8]}] real filename from header: '{stem}'")
                    except Exception as he:
                        log.debug(f"[{jid[:8]}] HEAD failed (non critico): {he}")
                    
                    async with c.stream("GET", video) as r:
                        if r.status_code == 404:
                            raise Exception(f"File non trovato (404). URL: {video[:80]}")
                        if r.status_code == 403:
                            raise Exception(f"Accesso negato (403)")
                        if r.status_code not in (200, 206):
                            raise Exception(f"HTTP {r.status_code}")
                        
                        total = int(r.headers.get("content-length", 0))
                        if total and total > MAX_UPLOAD_MB * 1024 * 1024:
                            raise HTTPException(413, f"File troppo grande: {total/1048576:.1f} MB > {MAX_UPLOAD_MB} MB")
                        
                        size = 0
                        last_t = 0
                        t_start = time.time()
                        async with aiofiles.open(vpath, "wb") as f:
                            async for ch in r.aiter_bytes(CHUNK_SIZE * 4):
                                await f.write(ch)
                                size += len(ch)
                                if size > MAX_UPLOAD_MB * 1024 * 1024:
                                    raise HTTPException(413, f"Max {MAX_UPLOAD_MB} MB")
                                now = time.time()
                                if now - last_t > 0.05:
                                    elapsed = now - t_start
                                    speed = size / elapsed / 1048576 if elapsed > 0 else 0
                                    pct = (size/total*98) if total else min(98, size/(1024*1024))
                                    if total:
                                        eta = (total - size) / (size/elapsed) if size > 0 else 0
                                        update(jid, step_progress=round(pct, 2),
                                               message=f"Download {size/1048576:.2f}/{total/1048576:.2f} MB · {speed:.2f} MB/s · ETA {eta:.1f}s")
                                    else:
                                        update(jid, step_progress=round(pct, 2),
                                               message=f"Download {size/1048576:.2f} MB · {speed:.2f} MB/s")
                                    last_t = now
                
                # SUCCESS uscita retry
                break
                
            except HTTPException:
                raise
            except Exception as e:
                last_error = e
                log.warning(f"[{jid[:8]}] download attempt {attempt+1}/3 failed: {e}")
                if attempt < 2:
                    update(jid, message=f"Retry {attempt+2}/3 dopo errore…")
                    await asyncio.sleep(2.0)
                    # Reset file
                    try: os.remove(vpath)
                    except: pass
                    size = 0
                else:
                    raise Exception(f"Download fallito dopo 3 tentativi: {last_error}")
        
        if size < MIN_FILE_BYTES:
            raise HTTPException(400, f"File troppo piccolo ({size} bytes)")
        
        update(jid, step_progress=99, message="Validazione…")
        await validate_video(vpath)
        log.info(f"[{jid[:8]}] download OK: {size/1048576:.2f} MB")
        update(jid, step=0, step_progress=100, message=f"Video pronto ({size/1048576:.2f} MB)",
               video_path=vpath, video_ready=True)
        return {"ok": True, "size_mb": round(size/1048576, 2), "name": j["original_name"]}
    except HTTPException as e:
        cleanup_paths([vpath])
        update(jid, error=str(e.detail), done=True)
        raise
    except Exception as e:
        log.exception(f"[{jid[:8]}] download failed")
        cleanup_paths([vpath])
        update(jid, error=str(e)[:200], done=True)
        raise HTTPException(500, str(e)[:200])


# ══════════════════════════════════════════════════════════════
# REST API — ONE-SHOT ENDPOINTS (per integrazione esterna)
# ══════════════════════════════════════════════════════════════
@app.get("/api/dub", tags=["REST API"])
async def api_dub_get(request: Request,
                      video: str = Query(..., description="URL diretto MP4 (http/https)"),
                      lang: str = Query("it", description="Codice lingua target (es. it, en, es)"),
                      gender: str = Query("Female", description="Female | Male"),
                      mode: str = Query("replace", description="replace | mix | advanced"),
                      dub_volume: float = 0.85,
                      orig_volume: float = 0.15,
                      fade: float = 1.0,
                      rate: str = "+0%",
                      pitch: str = "+0Hz",
                      volume: str = "+0%",
                      wait: bool = Query(False, description="True = blocca finché completato (max 30min)"),
                      timeout: int = Query(1800, description="Timeout in secondi se wait=true")):
    """
    REST GET — Avvia doppiaggio da URL.
    
    **Esempio sincrono (blocca):**
    ```
    GET /api/dub?video=https://example.com/video.mp4&lang=it&wait=true
    ```
    
    **Esempio asincrono (subito job_id, polla status):**
    ```
    GET /api/dub?video=https://example.com/video.mp4&lang=it
    → {"job_id": "abc...", "status_url": "..."}
    GET /status/abc...
    ```
    """
    # validazione URL
    video = video.strip()
    if video.startswith("blob:"):
        raise HTTPException(400, "blob: non supportato")
    if not re.match(r'^https?://', video, re.I):
        raise HTTPException(400, "URL deve iniziare con http/https")
    
    base = str(request.base_url).rstrip("/")
    
    # 1. Crea job
    jid = new_job()
    JOBS[jid]["params"] = make_params(lang, gender, mode, dub_volume, orig_volume, fade, rate, pitch, volume)
    update(jid, stage="upload", step=0, step_progress=0, message="Avvio…")
    asyncio.create_task(run_pipeline(jid, base))
    
    # 2. Avvia download (fire and forget se !wait, awaited se wait)
    async def do_download():
        try:
            uid = uuid.uuid4().hex
            vpath = os.path.join(TMP, f"dub_in_{uid}.mp4")
            async with httpx.AsyncClient(timeout=600, follow_redirects=True,
                                         headers={"User-Agent": "Mozilla/5.0"}) as c:
                async with c.stream("GET", video) as r:
                    if r.status_code not in (200, 206):
                        raise Exception(f"HTTP {r.status_code}")
                    total = int(r.headers.get("content-length", 0))
                    if total and total > MAX_UPLOAD_MB * 1024 * 1024:
                        raise Exception(f"Max {MAX_UPLOAD_MB} MB")
                    size = 0
                    last_t = 0
                    async with aiofiles.open(vpath, "wb") as f:
                        async for ch in r.aiter_bytes(CHUNK_SIZE * 4):
                            await f.write(ch); size += len(ch)
                            if size > MAX_UPLOAD_MB * 1024 * 1024:
                                raise Exception(f"Max {MAX_UPLOAD_MB} MB")
                            now = time.time()
                            if now - last_t > 0.05:
                                pct = (size/total*95) if total else min(95, size/(1024*1024))
                                update(jid, step_progress=round(pct, 2),
                                       message=f"Download {size/1048576:.1f} MB")
                                last_t = now
            if size < MIN_FILE_BYTES:
                raise Exception("File troppo piccolo")
            await validate_video(vpath)
            update(jid, step=0, step_progress=100, message="Video pronto",
                   video_path=vpath, video_ready=True)
        except Exception as e:
            log.error(f"[{jid[:8]}] api download fail: {e}")
            update(jid, error=f"Download fallito: {e}", done=True)
    
    if not wait:
        asyncio.create_task(do_download())
        return {
            "job_id": jid,
            "status_url": f"{base}/status/{jid}",
            "progress_url": f"{base}/progress/{jid}",
            "cancel_url": f"{base}/cancel/{jid}",
        }
    
    # Modalità sincrona
    await do_download()
    deadline = time.time() + timeout
    while time.time() < deadline:
        j = JOBS.get(jid)
        if not j:
            raise HTTPException(500, "Job perso")
        if j["done"]:
            if j.get("error"):
                raise HTTPException(500, j["error"])
            return {
                "job_id": jid,
                "status": "completed",
                **(j["result"] or {}),
            }
        await asyncio.sleep(1.0)
    raise HTTPException(504, f"Timeout dopo {timeout}s. Job ancora attivo: {jid}")


@app.post("/api/dub", tags=["REST API"])
async def api_dub_post(request: Request,
                       video: UploadFile = File(...),
                       lang: str = "it", gender: str = "Female", mode: str = "replace",
                       dub_volume: float = 0.85, orig_volume: float = 0.15, fade: float = 1.0,
                       rate: str = "+0%", pitch: str = "+0Hz", volume: str = "+0%",
                       wait: bool = False, timeout: int = 1800):
    """
    REST POST — Doppiaggio da file uploadato (multipart/form-data).
    
    **curl esempio:**
    ```
    curl -F "video=@video.mp4" "https://host/api/dub?lang=it&wait=true"
    ```
    """
    base = str(request.base_url).rstrip("/")
    jid = new_job()
    JOBS[jid]["params"] = make_params(lang, gender, mode, dub_volume, orig_volume, fade, rate, pitch, volume)
    update(jid, stage="upload", step=0, step_progress=0, message="Ricezione…")
    asyncio.create_task(run_pipeline(jid, base))
    
    uid = uuid.uuid4().hex
    vpath = os.path.join(TMP, f"dub_in_{uid}.mp4")
    size = 0
    try:
        async with aiofiles.open(vpath, "wb") as f:
            last_t = 0
            while True:
                c = await video.read(CHUNK_SIZE * 4)
                if not c: break
                await f.write(c); size += len(c)
                if size > MAX_UPLOAD_MB * 1024 * 1024:
                    raise HTTPException(413, f"Max {MAX_UPLOAD_MB} MB")
                now = time.time()
                if now - last_t > 0.05:
                    update(jid, step_progress=min(95, size/(MAX_UPLOAD_MB*1024*1024)*95),
                           message=f"Upload {size/1048576:.1f} MB")
                    last_t = now
        if size < MIN_FILE_BYTES:
            raise HTTPException(400, "File troppo piccolo")
        await validate_video(vpath)
        update(jid, step=0, step_progress=100, message="Video pronto",
               video_path=vpath, video_ready=True)
    except HTTPException as e:
        cleanup_paths([vpath])
        update(jid, error=str(e.detail), done=True)
        raise
    except Exception as e:
        cleanup_paths([vpath])
        update(jid, error=str(e), done=True)
        raise HTTPException(500, str(e))
    
    if not wait:
        return {
            "job_id": jid,
            "size_mb": round(size/1048576, 2),
            "status_url": f"{base}/status/{jid}",
            "progress_url": f"{base}/progress/{jid}",
        }
    
    deadline = time.time() + timeout
    while time.time() < deadline:
        j = JOBS.get(jid)
        if not j: raise HTTPException(500, "Job perso")
        if j["done"]:
            if j.get("error"):
                raise HTTPException(500, j["error"])
            return {"job_id": jid, "status": "completed", **(j["result"] or {})}
        await asyncio.sleep(1.0)
    raise HTTPException(504, f"Timeout dopo {timeout}s")


# ══════════════════════════════════════════════════════════════
# FILE SERVE & META
# ══════════════════════════════════════════════════════════════
@app.get("/file/{name}", tags=["Files"])
async def serve(name: str, dl: int = 0):
    path = os.path.join(PUBLIC, name)
    if not os.path.exists(path): raise HTTPException(404)
    mt = "video/mp4" if name.endswith(".mp4") else "text/plain; charset=utf-8"
    headers = {"Cache-Control": "public, max-age=3600"}
    if dl:
        return FileResponse(path, media_type=mt, filename=name, headers=headers)
    return FileResponse(path, media_type=mt, headers=headers)


@app.get("/languages", tags=["Meta"])
async def languages():
    return {"languages": [{"code":k, "name":v} for k,v in LANGUAGES.items()]}


@app.get("/voices", tags=["Meta"])
async def voices():
    return {"voices": [{"lang":l, "gender":g, "voice":v} for (l,g),v in VOICES.items()]}


@app.get("/health", tags=["Meta"])
async def health():
    return {"status":"ok", "version": "2.6.0",
            "jobs_active": sum(1 for j in JOBS.values() if not j["done"]),
            "jobs_total": len(JOBS),
            "max_upload_mb": MAX_UPLOAD_MB}


@app.get("/jobs", tags=["Jobs"])
async def list_jobs(active_only: bool = False):
    """Lista jobs con stato corrente."""
    out = []
    for jid, j in JOBS.items():
        if active_only and j["done"]: continue
        out.append({
            "id": jid, "stage": j["stage"], "progress": j["progress"],
            "step": j["step"], "done": j["done"], "error": j["error"],
            "created": j["created"],
        })
    return {"jobs": out, "count": len(out)}


# ══════════════════════════════════════════════════════════════
# UI
# ══════════════════════════════════════════════════════════════
@app.get("/", response_class=HTMLResponse, tags=["UI"])
async def home():
    opts = "".join(f'<option value="{k}">{k} — {v}</option>' for k, v in LANGUAGES.items())
    return UI.replace("__LANG_OPTIONS__", opts).replace("__MAX_MB__", str(MAX_UPLOAD_MB))


UI = r"""<!DOCTYPE html>
<html lang="it"><head><meta charset="UTF-8">
<title>DubFlow Web</title>
<meta name="viewport" content="width=device-width,initial-scale=1.0">
<script src="https://cdn.tailwindcss.com"></script>
<style>
body{background:radial-gradient(900px 600px at 10% -10%,rgba(139,92,246,.18),transparent 60%),radial-gradient(800px 500px at 100% 0%,rgba(56,189,248,.15),transparent 60%),#07090d;min-height:100vh;font-family:'Inter',system-ui,sans-serif;}
.glass{background:rgba(255,255,255,.04);backdrop-filter:blur(18px);border:1px solid rgba(255,255,255,.08);}
.drop{border:1.5px dashed rgba(255,255,255,.18);transition:.25s;}
.drop.drag{border-color:#a78bfa;background:rgba(167,139,250,.08);}
.btn{background:linear-gradient(90deg,#8b5cf6,#6366f1,#3b82f6);background-size:200% 100%;transition:.4s;}
.btn-download{display:flex;align-items:center;justify-content:center;gap:6px;white-space:nowrap;overflow:hidden;text-overflow:ellipsis;padding:10px 14px;min-width:0;max-width:100%;}
.btn-download span{overflow:hidden;text-overflow:ellipsis;white-space:nowrap;font-size:11.5px;font-family:'JetBrains Mono',monospace;}
.btn-download::before{content:'⬇';flex-shrink:0;font-size:14px;}
.dl-row{display:grid;grid-template-columns:1fr 1fr;gap:8px;}
.dl-row > a{min-width:0;}
.btn:hover:not(:disabled){background-position:100% 0;box-shadow:0 10px 40px rgba(139,92,246,.35);}
.btn:disabled{opacity:.5;cursor:not-allowed;}
.tab{transition:.25s;border:1px solid rgba(255,255,255,.08);}
.tab.active{background:linear-gradient(90deg,#8b5cf6,#6366f1);border-color:transparent;}
.bar-track{height:8px;background:rgba(255,255,255,.06);border:1px solid rgba(255,255,255,.08);border-radius:999px;overflow:hidden;}
.bar-fill{height:100%;width:0%;background:linear-gradient(90deg,#a78bfa,#6366f1,#3b82f6,#a78bfa);background-size:300% 100%;animation:flow 3s linear infinite;transition:width 80ms linear;border-radius:999px;box-shadow:0 0 20px rgba(139,92,246,.5);will-change:width;}
.bar-fill.step{background:linear-gradient(90deg,#22d3ee,#06b6d4,#0891b2,#22d3ee);background-size:300% 100%;}
.bar-fill.upload{background:linear-gradient(90deg,#10b981,#059669,#047857,#10b981);background-size:300% 100%;}
@keyframes flow{0%{background-position:0% 0}100%{background-position:300% 0}}
.code{font-family:'JetBrains Mono',monospace;font-size:11px;}
.label{letter-spacing:.12em;text-transform:uppercase;font-size:11px;color:#94a3b8;}
input,select{background:rgba(255,255,255,.05);border:1px solid rgba(255,255,255,.1);}
input:focus,select:focus{border-color:#a78bfa;box-shadow:0 0 0 3px rgba(167,139,250,.15);outline:none;}
.step-row{display:flex;align-items:flex-start;gap:10px;padding:12px;border-radius:10px;background:rgba(255,255,255,.02);border:1px solid rgba(255,255,255,.05);transition:background .3s, border-color .3s;}
.step-row.done-row{background:rgba(16,185,129,.04);border-color:rgba(16,185,129,.15);}
.step-row.active{background:rgba(139,92,246,.08);border-color:rgba(139,92,246,.3);}
.step-num{width:28px;height:28px;border-radius:999px;display:flex;align-items:center;justify-content:center;font-weight:600;font-size:12px;background:rgba(255,255,255,.05);border:1px solid rgba(255,255,255,.1);color:#94a3b8;flex-shrink:0;transition:.3s;margin-top:1px;}
.step-num.active{background:linear-gradient(90deg,#8b5cf6,#3b82f6);border-color:transparent;color:white;animation:pulse 2s infinite;}
.step-num.done{background:#10b981;border-color:transparent;color:white;}
@keyframes pulse{0%,100%{box-shadow:0 0 0 0 rgba(139,92,246,.5)}50%{box-shadow:0 0 0 8px rgba(139,92,246,0)}}
.toast{position:fixed;bottom:20px;right:20px;padding:14px 22px;border-radius:10px;color:white;font-size:13px;animation:slideIn .3s;z-index:100;max-width:420px;box-shadow:0 10px 40px rgba(0,0,0,.4);}
.toast.err{background:rgba(239,68,68,.95);} .toast.ok{background:rgba(16,185,129,.95);} .toast.info{background:rgba(59,130,246,.95);}
@keyframes slideIn{from{transform:translateX(100%);opacity:0}to{transform:translateX(0);opacity:1}}
.hint{font-size:11px;color:#64748b;margin-top:4px;}
.step-msg{font-size:10.5px;color:#94a3b8;margin-top:6px;font-family:'JetBrains Mono',monospace;line-height:1.3;min-height:13px;letter-spacing:.01em;}
.step-row.active .step-msg{color:#c4b5fd;}
.step-row.done-row .step-msg{color:#6ee7b7;}
.speed{font-size:10px;color:#10b981;font-family:'JetBrains Mono',monospace;}
.heartbeat{display:inline-block;width:6px;height:6px;border-radius:50%;background:#10b981;margin-left:6px;animation:beat 1s infinite;}
@keyframes beat{0%,100%{opacity:.3}50%{opacity:1;box-shadow:0 0 8px #10b981}}
</style></head><body class="text-slate-100">

<div class="max-w-5xl mx-auto pt-10 px-6 pb-20">
<header class="mb-8">
<p class="label mb-2">Engine v2.6.0 — Real-time + REST API</p>
<h1 class="text-4xl md:text-5xl font-semibold tracking-tight">DubFlow <span class="text-slate-400 font-light">Web</span></h1>
<p class="text-slate-400 mt-2 text-sm">Pipeline: upload · estrazione · traduzione · TTS · dub</p>
</header>

<div class="flex gap-2 mb-5">
<button onclick="tab('file')" id="t-file" class="tab active flex-1 py-2.5 rounded-xl text-sm font-medium">Upload File</button>
<button onclick="tab('url')" id="t-url" class="tab flex-1 py-2.5 rounded-xl text-sm font-medium">URL Remoto</button>
</div>

<div class="grid md:grid-cols-2 gap-5">
<div class="glass rounded-2xl p-6 space-y-4">

<div id="pane-file">
<div id="drop" class="drop rounded-xl p-8 text-center cursor-pointer">
<p class="label mb-1">Video</p>
<p class="text-slate-300 text-sm">Trascina o clicca</p>
<p id="fname" class="text-xs text-slate-500 mt-2">Nessun file (max __MAX_MB__ MB)</p>
<input id="vfile" type="file" accept="video/*" class="hidden">
</div>
</div>

<div id="pane-url" class="hidden">
<p class="label mb-2">URL Video diretto</p>
<input id="vurl" type="text" placeholder="https://esempio.com/video.mp4" class="w-full px-4 py-3 rounded-xl text-sm">
<p class="hint">Solo HTTP/HTTPS. Niente <code class="text-violet-400">blob:</code></p>
</div>

<div class="grid grid-cols-2 gap-3">
<div><p class="label mb-2">Lingua</p>
<select id="lang" class="w-full px-3 py-2.5 rounded-xl text-sm">__LANG_OPTIONS__</select></div>
<div><p class="label mb-2">Voce</p>
<select id="gender" class="w-full px-3 py-2.5 rounded-xl text-sm">
<option value="Female">Femminile</option><option value="Male">Maschile</option></select></div>
</div>

<div><p class="label mb-2">Modalità Dub</p>
<select id="mode" class="w-full px-3 py-2.5 rounded-xl text-sm">
<option value="replace">Replace</option>
<option value="mix">Mix</option>
<option value="advanced">Advanced</option>
</select></div>

<details class="text-sm">
<summary class="cursor-pointer text-slate-400 hover:text-slate-200 select-none">Parametri avanzati</summary>
<div class="mt-3 space-y-3">
<div class="grid grid-cols-3 gap-2">
<div><p class="label mb-1">Rate</p><input id="rate" value="+0%" class="w-full px-2 py-2 rounded-lg text-xs"></div>
<div><p class="label mb-1">Pitch</p><input id="pitch" value="+0Hz" class="w-full px-2 py-2 rounded-lg text-xs"></div>
<div><p class="label mb-1">Vol TTS</p><input id="vol" value="+0%" class="w-full px-2 py-2 rounded-lg text-xs"></div>
</div>
<div class="grid grid-cols-3 gap-2">
<div><p class="label mb-1">Dub vol</p><input id="dubv" type="number" value="0.85" step="0.05" min="0" max="1" class="w-full px-2 py-2 rounded-lg text-xs"></div>
<div><p class="label mb-1">Orig vol</p><input id="origv" type="number" value="0.15" step="0.05" min="0" max="1" class="w-full px-2 py-2 rounded-lg text-xs"></div>
<div><p class="label mb-1">Fade</p><input id="fade" type="number" value="1.0" step="0.5" min="0" max="5" class="w-full px-2 py-2 rounded-lg text-xs"></div>
</div>
</div>
</details>

<button onclick="start()" id="btnStart" class="btn w-full py-3.5 rounded-xl font-semibold">Avvia doppiaggio</button>
<button onclick="cancel()" id="btnCancel" class="hidden w-full py-2.5 rounded-xl font-semibold bg-red-500/20 border border-red-500/30 text-red-300 hover:bg-red-500/30">Annulla</button>

</div>

<div class="glass rounded-2xl p-6 space-y-4">

<div>
<div class="flex justify-between text-xs text-slate-400 mb-2">
<span><span id="globalStage">In attesa</span><span id="hb" class="heartbeat" style="display:none"></span></span>
<span id="globalPct" class="code text-violet-300">0.00 %</span>
</div>
<div class="bar-track"><div id="globalBar" class="bar-fill"></div></div>
</div>

<div class="space-y-2">
<div class="step-row" id="row0">
<div class="step-num" id="num0">↑</div>
<div class="flex-1 min-w-0">
<div class="flex justify-between items-center"><p class="text-sm font-medium">Upload</p><span id="s0pct" class="code text-violet-300">—</span></div>
<div class="bar-track mt-1.5"><div id="s0bar" class="bar-fill upload"></div></div>
<p id="s0msg" class="step-msg">—</p>
</div>
</div>

<div class="step-row" id="row1">
<div class="step-num" id="num1">1</div>
<div class="flex-1 min-w-0">
<div class="flex justify-between items-center"><p class="text-sm font-medium">Extract SRT</p><span id="s1pct" class="code text-violet-300">—</span></div>
<div class="bar-track mt-1.5"><div id="s1bar" class="bar-fill step"></div></div>
<p id="s1msg" class="step-msg">—</p>
</div>
</div>

<div class="step-row" id="row2">
<div class="step-num" id="num2">2</div>
<div class="flex-1 min-w-0">
<div class="flex justify-between items-center"><p class="text-sm font-medium">Translate</p><span id="s2pct" class="code text-violet-300">—</span></div>
<div class="bar-track mt-1.5"><div id="s2bar" class="bar-fill step"></div></div>
<p id="s2msg" class="step-msg">—</p>
</div>
</div>

<div class="step-row" id="row3">
<div class="step-num" id="num3">3</div>
<div class="flex-1 min-w-0">
<div class="flex justify-between items-center"><p class="text-sm font-medium">Synced TTS</p><span id="s3pct" class="code text-violet-300">—</span></div>
<div class="bar-track mt-1.5"><div id="s3bar" class="bar-fill step"></div></div>
<p id="s3msg" class="step-msg">—</p>
</div>
</div>

<div class="step-row" id="row4">
<div class="step-num" id="num4">4</div>
<div class="flex-1 min-w-0">
<div class="flex justify-between items-center"><p class="text-sm font-medium">Dub Video</p><span id="s4pct" class="code text-violet-300">—</span></div>
<div class="bar-track mt-1.5"><div id="s4bar" class="bar-fill step"></div></div>
<p id="s4msg" class="step-msg">—</p>
</div>
</div>
</div>

<p id="msg" class="text-xs text-slate-500 code break-all min-h-[16px]"></p>

<details class="text-xs">
<summary class="cursor-pointer text-slate-500 hover:text-violet-300 select-none">🔍 Debug console (apri F12 per dettagli)</summary>
<pre id="debugLog" class="mt-2 p-2 bg-black/40 rounded text-[10px] text-emerald-300 max-h-32 overflow-y-auto code"></pre>
</details>

<div id="result" class="hidden space-y-3 pt-3 border-t border-white/5">
<video id="player" controls class="w-full rounded-xl border border-white/10"></video>
<div class="dl-row text-xs">
<a id="dlVideo" class="btn btn-download rounded-lg font-semibold" title="Scarica video doppiato"><span>Video MP4</span></a>
<a id="dlSrt" target="_blank" class="btn-download rounded-lg font-semibold bg-white/5 border border-white/10 hover:bg-white/10" title="Scarica sottotitoli tradotti"><span>SRT tradotto</span></a>
</div>
</div>

</div>
</div>

<div class="glass rounded-2xl p-5 mt-5 text-xs">
<p class="label mb-3">REST API — Esempi</p>
<div class="space-y-2 text-slate-400 code">
<div><span class="text-emerald-400">GET</span> <code class="text-violet-300">/api/dub?video=URL&lang=it&wait=true</code> — sincrono</div>
<div><span class="text-emerald-400">GET</span> <code class="text-violet-300">/api/dub?video=URL&lang=it</code> → restituisce <code>job_id</code></div>
<div><span class="text-blue-400">POST</span> <code class="text-violet-300">/api/dub</code> (multipart: video=@file.mp4)</div>
<div><span class="text-emerald-400">GET</span> <code class="text-violet-300">/status/{job_id}</code> — polling</div>
<div><span class="text-emerald-400">GET</span> <code class="text-violet-300">/progress/{job_id}</code> — SSE stream</div>
</div>
</div>

<footer class="text-center mt-8 text-slate-500 text-xs space-x-5">
<a href="/docs" class="hover:text-violet-300">Swagger</a>
<a href="/redoc" class="hover:text-violet-300">ReDoc</a>
<a href="/health" class="hover:text-violet-300">Health</a>
<a href="/languages" class="hover:text-violet-300">Lingue</a>
</footer>
</div>

<script>
const $ = id => document.getElementById(id);
const MAX_MB = __MAX_MB__;
let currentJob = null, currentES = null, currentXHR = null;
let uploadStartTime = 0;

// Sistema tween: interpola tra valore corrente e target ad ogni frame
const targetBars = {global: 0, s0: 0, s1: 0, s2: 0, s3: 0, s4: 0};
const currentBars = {global: 0, s0: 0, s1: 0, s2: 0, s3: 0, s4: 0};

function tweenStep(){
 const SPEED = 0.35; // 35% del delta = più reattivo
 for(const k of Object.keys(targetBars)){
  const target = targetBars[k];
  const current = currentBars[k];
  const delta = target - current;
  if(Math.abs(delta) > 0.01){
   currentBars[k] = current + delta * SPEED;
  } else {
   currentBars[k] = target;
  }
  const val = currentBars[k];
  if(k === 'global'){
   $('globalBar').style.width = val.toFixed(2)+'%';
   $('globalPct').textContent = val.toFixed(2)+' %';
  } else {
   const bar = document.getElementById(k+'bar');
   const pct = document.getElementById(k+'pct');
   if(bar) bar.style.width = val.toFixed(2)+'%';
   if(pct) pct.textContent = val.toFixed(2)+'%';
  }
 }
 requestAnimationFrame(tweenStep);
}
requestAnimationFrame(tweenStep);

function log(...a){
 console.log('[DubFlow]', ...a);
 const dl = document.getElementById('debugLog');
 if(dl){
  const line = '['+new Date().toLocaleTimeString()+'] '+a.map(x=>typeof x==='object'?JSON.stringify(x):String(x)).join(' ');
  dl.textContent = (line+'\n'+dl.textContent).slice(0, 4000);
 }
}

// Intercetta anche console.log/warn/error per debug panel
(function(){
 const _log = console.log;
 const _warn = console.warn;
 const _err = console.error;
 const append = (level, args) => {
  const dl = document.getElementById('debugLog');
  if(!dl) return;
  const line = '['+new Date().toLocaleTimeString()+'] ['+level+'] '+Array.from(args).map(x=>typeof x==='object'?JSON.stringify(x):String(x)).join(' ');
  dl.textContent = (line+'\n'+dl.textContent).slice(0, 4000);
 };
 console.log = function(...a){ _log.apply(console,a); append('LOG',a); };
 console.warn = function(...a){ _warn.apply(console,a); append('WRN',a); };
 console.error = function(...a){ _err.apply(console,a); append('ERR',a); };
})();

function toast(msg, type='err', dur=5000){
 const t=document.createElement('div');t.className='toast '+type;t.textContent=msg;
 document.body.appendChild(t);
 setTimeout(()=>{t.style.opacity='0';t.style.transition='opacity .3s';setTimeout(()=>t.remove(),300);},dur);
}

function tab(n){
 ['file','url'].forEach(t=>{$('pane-'+t).classList.toggle('hidden', t!==n);$('t-'+t).classList.toggle('active', t===n);});
}

(function(){
 const z=$('drop'),i=$('vfile');
 z.onclick=()=>i.click();
 z.ondragover=e=>{e.preventDefault();z.classList.add('drag');};
 z.ondragleave=()=>z.classList.remove('drag');
 z.ondrop=e=>{e.preventDefault();z.classList.remove('drag');
  if(e.dataTransfer.files.length){i.files=e.dataTransfer.files;showFile(i.files[0]);}};
 i.onchange=()=>{if(i.files[0]) showFile(i.files[0]);};
})();

function showFile(f){
 const mb=(f.size/1048576).toFixed(1);
 $('fname').textContent = `${f.name} — ${mb} MB`;
 if(f.size > MAX_MB*1048576){$('fname').classList.add('text-red-400');}
 else {$('fname').classList.remove('text-red-400');$('fname').classList.add('text-emerald-400');}
}

function setBar(id, pct){
 const p=Math.max(0,Math.min(100,pct));
 $(id+'bar').style.width=p.toFixed(2)+'%';
 $(id+'pct').textContent=p.toFixed(1)+'%';
}

function setStep(n, status){
 const el=$('num'+n), row=$('row'+n);
 el.classList.remove('active','done');row.classList.remove('active');
 if(status==='active'){el.classList.add('active');row.classList.add('active');}
 if(status==='done'){el.classList.add('done'); el.textContent=n===0?'↑':'✓';}
 else el.textContent=n===0?'↑':String(n);
}

function resetUI(){
 for(let i=0;i<=4;i++){
  setStep(i,'idle');
  targetBars['s'+i]=0;
  currentBars['s'+i]=0;
  const bar=document.getElementById('s'+i+'bar');
  const pct=document.getElementById('s'+i+'pct');
  const msg=document.getElementById('s'+i+'msg');
  const row=document.getElementById('row'+i);
  if(bar) bar.style.width='0%';
  if(pct) pct.textContent='—';
  if(msg){msg.textContent='—'; delete msg.dataset.locked;}
  if(row) row.classList.remove('active', 'done-row');
 }
 targetBars['global']=0;
 currentBars['global']=0;
 $('globalBar').style.width='0%';
 $('globalPct').textContent='0.00 %';
 $('globalStage').textContent='Avvio';
 $('result').classList.add('hidden');
 $('msg').textContent='';
 $('hb').style.display='none';
}

function track(jid){
 log('tracking', jid);
 currentJob=jid;
 $('btnCancel').classList.remove('hidden');$('btnStart').disabled=true;
 if(currentES){try{currentES.close();}catch{}}
 
 let lastSSETime = Date.now();
 let pollFallbackTimer = null;
 let sseConnected = false;
 
 const handleUpdate = (d, source) => {
  if(d.error){
   toast('Errore: '+d.error,'err',10000);
   $('btnCancel').classList.add('hidden');$('btnStart').disabled=false;
   $('hb').style.display='none';
   if(currentES) try{currentES.close();}catch{}
   if(pollFallbackTimer) clearInterval(pollFallbackTimer);
   return false;
  }
  
  // DEBUG: log ogni messaggio
  console.log(`[${source}] step=${d.step} progress=${d.progress} step_progress=${d.step_progress} msg="${d.message}"`);
  
  const gp = Number(d.progress) || 0;
  const curStep = d.step !== undefined ? Number(d.step) : 0;
  const curStepProg = Number(d.step_progress) || 0;
  const curMsg = d.message || '';
  
  targetBars['global'] = gp;
  $('globalStage').textContent = d.stage || '';
  $('msg').textContent = `[${source}] step=${curStep} ${curStepProg.toFixed(2)}% · ${curMsg}`;
  
  for(let i=0; i<=4; i++){
   const msgEl = document.getElementById('s'+i+'msg');
   const rowEl = document.getElementById('row'+i);
   
   if(curStep > i){
    setStep(i, 'done');
    targetBars['s'+i] = 100;
    currentBars['s'+i] = 100;
    if(rowEl){rowEl.classList.add('done-row'); rowEl.classList.remove('active');}
    if(msgEl && !msgEl.dataset.locked){
     msgEl.textContent = '✓ completato';
     msgEl.dataset.locked = '1';
    }
   }
   else if(curStep === i){
    setStep(i, 'active');
    targetBars['s'+i] = curStepProg;
    if(rowEl){rowEl.classList.add('active'); rowEl.classList.remove('done-row');}
    if(msgEl){
     msgEl.textContent = curMsg || '...';
     delete msgEl.dataset.locked;
    }
   }
   else {
    setStep(i, 'idle');
    targetBars['s'+i] = 0;
    currentBars['s'+i] = 0;
    if(rowEl){rowEl.classList.remove('active', 'done-row');}
    if(msgEl){
     msgEl.textContent = '—';
     delete msgEl.dataset.locked;
    }
   }
  }
  
  if(d.done && d.result){
   for(let i=0;i<=4;i++){
    setStep(i,'done');
    targetBars['s'+i]=100;
    currentBars['s'+i]=100;
    const bar=document.getElementById('s'+i+'bar');
    const pct=document.getElementById('s'+i+'pct');
    const msg=document.getElementById('s'+i+'msg');
    const row=document.getElementById('row'+i);
    if(bar) bar.style.width='100%';
    if(pct) pct.textContent='100.00%';
    if(msg) msg.textContent='✓ completato';
    if(row){row.classList.add('done-row'); row.classList.remove('active');}
   }
   targetBars['global']=100;
   currentBars['global']=100;
   $('globalBar').style.width='100%';
   $('globalPct').textContent='100.00 %';
   $('globalStage').textContent='completato';
   $('hb').style.display='none';
   $('result').classList.remove('hidden');
   $('player').src=d.result.video_url;
   $('dlVideo').href=d.result.video_download;
   if(d.result.video_filename){
    $('dlVideo').setAttribute('download', d.result.video_filename);
    $('dlVideo').textContent = '⬇ ' + d.result.video_filename;
   }
   $('dlSrt').href=d.result.srt_translated;
   $('btnCancel').classList.add('hidden');
   $('btnStart').disabled=false;
   toast('Completato in '+d.result.elapsed_sec+'s','ok');
   if(currentES) try{currentES.close();}catch{}
   if(pollFallbackTimer) clearInterval(pollFallbackTimer);
   return false;
  }
  return true;
 };
 
 // POLLING FALLBACK — si attiva se SSE non riceve nulla per >3s
 const startPollingFallback = () => {
  if(pollFallbackTimer) return;
  console.warn('[FALLBACK] Starting polling /status (SSE silent)');
  $('msg').textContent = '[FALLBACK POLLING ATTIVO] SSE non riceve eventi…';
  pollFallbackTimer = setInterval(async () => {
   if(currentJob !== jid){
    clearInterval(pollFallbackTimer);
    return;
   }
   try {
    const r = await fetch('/status/'+jid);
    if(!r.ok) return;
    const d = await r.json();
    if(!handleUpdate(d, 'POLL')){
     clearInterval(pollFallbackTimer);
    }
   } catch(e) {
    console.error('polling error', e);
   }
  }, 500);
 };
 
 // SSE CONNECTION
 const connectSSE = () => {
  console.log('[SSE] connecting…');
  const es = new EventSource('/progress/' + jid);
  currentES = es;
  $('hb').style.display = 'inline-block';
  
  es.onopen = () => {
   console.log('[SSE] connected');
   sseConnected = true;
   lastSSETime = Date.now();
  };
  
  es.onmessage = e => {
   lastSSETime = Date.now();
   sseConnected = true;
   if(pollFallbackTimer){
    console.log('[SSE] recovered, stopping polling fallback');
    clearInterval(pollFallbackTimer);
    pollFallbackTimer = null;
   }
   let d;
   try{ d = JSON.parse(e.data); }
   catch(err){ console.error('SSE parse error', err, e.data); return; }
   if(!handleUpdate(d, 'SSE')){
    es.close();
   }
  };
  
  es.onerror = e => {
   console.warn('[SSE] error', e, 'state=', es.readyState);
   $('hb').style.display='none';
   sseConnected = false;
   // Non chiudere subito, EventSource fa auto-reconnect
   // Ma se rimaniamo disconnessi >3s, attiva polling
   setTimeout(() => {
    if(!sseConnected && currentJob === jid){
     startPollingFallback();
    }
   }, 3000);
  };
 };
 
 // Watchdog: se SSE non riceve nulla per 3s, attiva polling
 const watchdog = setInterval(() => {
  if(currentJob !== jid){
   clearInterval(watchdog);
   return;
  }
  const silenceTime = (Date.now() - lastSSETime) / 1000;
  if(silenceTime > 3 && !pollFallbackTimer){
   console.warn(`[WATCHDOG] SSE silent for ${silenceTime.toFixed(1)}s, starting polling`);
   startPollingFallback();
  }
 }, 1000);
 
 connectSSE();
}

function getParams(){
 return {
  lang: $('lang').value, gender: $('gender').value, mode: $('mode').value,
  dub_volume: $('dubv').value, orig_volume: $('origv').value, fade: $('fade').value,
  rate: $('rate').value, pitch: $('pitch').value, volume: $('vol').value,
 };
}

async function parseErr(r){
 try{const j=await r.json();return j.detail||j.error||JSON.stringify(j);}
 catch{return (await r.text()).slice(0,200)||`HTTP ${r.status}`;}
}

function uploadFile(jid, file){
 return new Promise((resolve,reject)=>{
  const xhr=new XMLHttpRequest();
  currentXHR=xhr;uploadStartTime=Date.now();
  xhr.upload.onprogress=e=>{
   if(!e.lengthComputable) return;
   const pct=(e.loaded/e.total)*100;
   targetBars['s0']=pct;
   setStep(0,'active');
   const elapsed=(Date.now()-uploadStartTime)/1000;
   const speed=e.loaded/elapsed/1048576;
   const eta=e.loaded>0 ? (e.total-e.loaded)/(e.loaded/elapsed) : 0;
   const mb=(e.loaded/1048576).toFixed(2);
   const tot=(e.total/1048576).toFixed(2);
   const msg = `${mb}/${tot} MB · ${speed.toFixed(2)} MB/s · ETA ${eta.toFixed(1)}s`;
   const s0msg = document.getElementById('s0msg');
   if(s0msg){s0msg.textContent = msg; delete s0msg.dataset.locked;}
   const rowEl = document.getElementById('row0');
   if(rowEl) rowEl.classList.add('active');
  };
  xhr.upload.onload=()=>{
   const s0msg = document.getElementById('s0msg');
   if(s0msg) s0msg.textContent = 'upload completato, attesa server…';
  };
  xhr.onload=()=>{
   currentXHR=null;
   if(xhr.status>=200 && xhr.status<300){try{resolve(JSON.parse(xhr.responseText));}catch{resolve({});}}
   else {try{const j=JSON.parse(xhr.responseText);reject(new Error(j.detail||xhr.statusText));}catch{reject(new Error(xhr.responseText||xhr.statusText));}}
  };
  xhr.onerror=()=>{currentXHR=null;reject(new Error('Errore rete'));};
  xhr.onabort=()=>{currentXHR=null;reject(new Error('Annullato'));};
  xhr.ontimeout=()=>{currentXHR=null;reject(new Error('Timeout'));};
  xhr.timeout=30*60*1000;
  xhr.open('POST',`/job/${jid}/upload`);
  const fd=new FormData(); fd.append('video',file);
  xhr.send(fd);
 });
}

async function start(){
 const isFile = !$('pane-file').classList.contains('hidden');
 const p = getParams();
 if(isFile){
  const f=$('vfile').files[0];
  if(!f){toast('Seleziona video');return;}
  if(f.size<1000){toast('File troppo piccolo');return;}
  if(f.size>MAX_MB*1048576){toast(`Max ${MAX_MB} MB`);return;}
 } else {
  const u=$('vurl').value.trim();
  if(!u){toast('Inserisci URL');return;}
  if(u.startsWith('blob:')){toast('blob: non supportato','err',9000);return;}
  if(!/^https?:\/\//i.test(u)){toast('URL http/https');return;}
 }
 resetUI();$('btnStart').disabled=true;
 try {
  const qs=new URLSearchParams(p).toString();
  const r=await fetch('/job/create?'+qs,{method:'POST'});
  if(!r.ok){toast('Errore: '+await parseErr(r));$('btnStart').disabled=false;return;}
  const {job_id}=await r.json();
  track(job_id);
  await new Promise(r=>setTimeout(r,200));
  if(isFile){
   try { await uploadFile(job_id, $('vfile').files[0]); }
   catch(e){ toast('Upload fallito: '+e.message,'err',8000); return; }
  } else {
   const u=$('vurl').value.trim();
   setStep(0,'active');
   const ur=await fetch(`/job/${job_id}/url?video=${encodeURIComponent(u)}`,{method:'POST'});
   if(!ur.ok){toast('Download fallito: '+await parseErr(ur),'err',8000);return;}
  }
 } catch(e){
  toast('Errore: '+e.message,'err');$('btnStart').disabled=false;
 }
}

async function cancel(){
 if(currentXHR){try{currentXHR.abort();}catch{}}
 if(currentJob){try{await fetch('/cancel/'+currentJob,{method:'POST'});}catch{} toast('Annullato','info');}
 $('btnCancel').classList.add('hidden');$('btnStart').disabled=false;
}
</script>
</body></html>"""


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