""" 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}[^>]*>(.*?)', 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 ', 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*-->[^"\']*)["\']', 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'' for k, v in LANGUAGES.items()) return UI.replace("__LANG_OPTIONS__", opts).replace("__MAX_MB__", str(MAX_UPLOAD_MB)) UI = r""" DubFlow Web

Engine v2.6.0 — Real-time + REST API

DubFlow Web

Pipeline: upload · estrazione · traduzione · TTS · dub

Video

Trascina o clicca

Nessun file (max __MAX_MB__ MB)

Lingua

Voce

Modalità Dub

Parametri avanzati

Rate

Pitch

Vol TTS

Dub vol

Orig vol

Fade

In attesa 0.00 %
↑

Upload

—

—

1

Extract SRT

—

—

2

Translate

—

—

3

Synced TTS

—

—

4

Dub Video

—

—

🔍 Debug console (apri F12 per dettagli) 














REST API — Esempi




GET /api/dub?video=URL&lang=it&wait=true — sincrono


GET /api/dub?video=URL&lang=it → restituisce job_id


POST /api/dub (multipart: video=@file.mp4)


GET /status/{job_id} — polling


GET /progress/{job_id} — SSE stream











"""


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