"""DubLab — short-video dubbing website. Upload video -> dubbed video."""
import json
import os
import re
import shutil
import subprocess
import sys
import threading
import time
import urllib.request
import uuid
from collections import deque
from pathlib import Path
from urllib.parse import urlparse
from flask import Flask, jsonify, render_template, request, send_file
from alerts import quota_alert, quota_ok
from dub import (blur_region, build_dub_track, cleanup_segments, configure_tts,
extract_audio, has_audio_stream, has_video_stream,
merge_segments, mux, probe_duration, probe_size, set_engine,
set_nara_voice, transcribe, valid_nara_voice, write_ass,
write_srt)
from gemini import translate_lines
from luna import translate_lines as luna_translate_lines
from gemini_tts import QuotaExhausted
# Gemini TTS key rotation (same 5-key pool style as the SRT site).
# GEMINI_TTS_API_KEYS is comma-separated; falls back to the single GEMINI_API_KEY.
_tts_keys = [k.strip() for k in
os.environ.get("GEMINI_TTS_API_KEYS",
os.environ.get("GEMINI_API_KEY", "")).split(",")
if k.strip()]
configure_tts(_tts_keys)
BASE = Path(__file__).parent
JOBS = BASE / "jobs"
JOBS.mkdir(exist_ok=True)
# Link-mode blur marking downloads the video here first so the user can
# pick a frame and draw the blur box; from-url then moves the file into
# the job instead of downloading twice. Swept after an hour.
STAGE = BASE / "stage"
STAGE.mkdir(exist_ok=True)
STAGE_TTL = 3600
MAX_DURATION = 600 # seconds (raised 5→10 min 2026-11-07, his order)
MAX_MB = 250
TTL = 24 * 3600 # keep results a day
# ---- abuse protection (public site) ----
MAX_QUEUE = 15 # max jobs waiting in line
UPLOADS_PER_HOUR = 10 # per client IP
_ip_hits = {}
_ip_lock = threading.Lock()
# ---- maintenance mode (TTS outage) ----
# Turn True to pause new dub jobs; the home page shows the maintenance image.
MAINTENANCE = False
MAINTENANCE_MSG = ("🔧 DubLab ခေတ္တ ရပ်ထားပါတယ် — Microsoft အသံထုတ် service "
"ပျက်နေလို့ ပြင်ဆင်နေပါတယ်။ ပြန်ကောင်းတာနဲ့ အကြောင်းကြားမယ် 🙏")
# ---- reviews ----
REVIEWS = BASE / "reviews.jsonl"
ADMIN_KEY = os.environ.get("DUBLAB_ADMIN_KEY", "")
# ---- usage counter (admin panel only) ----
# Job dirs are swept after a day, so without this append-only log the
# totals would be lost: one line per job created / done / failed.
USAGE = BASE / "usage.jsonl"
_usage_lock = threading.Lock()
VOICES = {
"en2mm": {"female": "my-MM-NilarNeural", "male": "my-MM-ThihaNeural"},
"mm2en": {"female": "en-US-AvaNeural", "male": "en-US-AndrewNeural"},
}
STAGE_LABEL = {
"queue": "တန်းစီနေတယ်",
"download": "ဗီဒီယို ဆွဲချနေတယ်",
"blur": "မူရင်း စာတန်း ဖုံးနေတယ်",
"audio": "အသံထုတ်ယူနေတယ်",
"transcribe": "စကားပြောမှတ်နေတယ်",
"translate": "ဘာသာပြန်နေတယ်",
"voice": "အသံဖော်နေတယ်",
"mux": "ဗီဒီယိုပေါင်းနေတယ်",
"done": "ပြီးပြီ",
}
# ---- video link support (yt-dlp) ----
# Instead of uploading a file, users can paste a link from a social site.
# yt-dlp lives in the same venv as this app.
YTDLP = Path(sys.executable).parent / "yt-dlp"
LINK_HOSTS = ("youtube.com", "youtu.be", "tiktok.com", "instagram.com",
"facebook.com", "fb.watch", "xiaohongshu.com", "xhslink.com",
"rednote.com", "twitter.com", "x.com", "vimeo.com",
"dailymotion.com", "bilibili.com")
# best mp4 up to 720p; the /b fallback covers sites with combined formats only
LINK_FMT = "bv*[height<=720]+ba/b[height<=720]/b"
def _host_ok(host):
host = (host or "").lower()
return any(host == h or host.endswith("." + h) for h in LINK_HOSTS)
def _link_log(url, stage, msg):
"""Keep link failures on disk so a bad link can be diagnosed later."""
try:
line = f"{time.strftime('%F %T')} [{stage}] {url} :: {msg}\n"
with open(BASE / "link_check.log", "a") as fh:
fh.write(line.replace("\n", " ")[:600] + "\n")
except OSError:
pass
SHORT_HOSTS = ("xhslink.com", "fb.watch", "vm.tiktok.com", "vt.tiktok.com")
_MOBILE_UA = ("Mozilla/5.0 (iPhone; CPU iPhone OS 18_7 like Mac OS X) "
"AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 "
"Mobile/15E148 Safari/604.1")
def resolve_link(url):
"""Follow a share short-link to its real URL.
RedNote's xhslink lands on rednote.com, which yt-dlp can't read and
which bounces anonymous visitors to a login page. The same URL on
xiaohongshu.com (with its xsec_token kept) works, so the host is
rewritten after resolving. Other short links pass through unchanged
apart from the redirect itself.
"""
host = (urlparse(url).hostname or "").lower()
if host not in SHORT_HOSTS:
return url
req = urllib.request.Request(url, headers={"User-Agent": _MOBILE_UA})
try:
with urllib.request.urlopen(req, timeout=20) as r:
final = r.geturl()
except Exception: # noqa: BLE001 - fall back to the short link itself
return url
parts = urlparse(final)
if (parts.hostname or "").lower() in ("rednote.com", "www.rednote.com"):
final = parts._replace(netloc="www.xiaohongshu.com").geturl()
return final
def check_link(url):
"""Check a pasted link looks safe to hand to yt-dlp. Returns the url.
Share buttons often copy a whole sentence with the link inside, so the
first http(s) URL in the pasted text is what counts. RedNote's
international domain (rednote.com) is the same site as xiaohongshu.com
but yt-dlp only knows the .com name, so it is rewritten here.
"""
text = (url or "").strip()
m = re.search(r"https?://[^\s\"'<>]+", text)
url = (m.group(0).rstrip(").,，。") if m else "")
if len(url) > 2000 or not url:
raise ValueError("link မှားနေတယ် — http/https link ပဲ ထည့်ပါ")
parts = urlparse(url)
host = (parts.hostname or "").lower()
if host in ("rednote.com", "www.rednote.com"):
url = parts._replace(netloc="www.xiaohongshu.com").geturl()
host = "www.xiaohongshu.com"
if not _host_ok(host):
raise ValueError("ဒီ site က link လက်မခံဘူး — YouTube / TikTok / RedNote / "
"Instagram / Facebook link ပဲ ရတယ်")
url = resolve_link(url)
if not _host_ok(urlparse(url).hostname):
raise ValueError("ဒီ site က link လက်မခံဘူး — YouTube / TikTok / RedNote / "
"Instagram / Facebook link ပဲ ရတယ်")
return url
def link_err(stderr):
"""Turn yt-dlp errors into short Burmese messages."""
s = (stderr or "").lower()
if "xiaohongshu" in s or "rednote" in s:
return ("RedNote link ကို server က ဖွင့်မရဘူး — app ထဲက Share → "
"Copy link နဲ့ ရတဲ့ link ကို အပြည့်အစုံ ပြန် paste လုပ်ကြည့်ပါ")
if ("private" in s or "login" in s or "cookies" in s or "sign in" in s
or "log in" in s):
return ("ဒီ video က private / login လိုတယ် — video file ကို "
"တိုက်ရိုက် တင်ပေးပါ")
if "no video" in s:
return "ဒီ link မှာ video မတွေ့ဘူး"
return "link က video ကို ဖတ်မရဘူး — link မှန်ရဲ့လား စစ်ပြီး ပြန်စမ်းပါ"
def ytdlp_info(url):
"""Fetch link metadata with yt-dlp (no download)."""
p = subprocess.run([str(YTDLP), "--no-playlist", "--no-warnings",
"--impersonate", "chrome", "-J", url],
capture_output=True, text=True, timeout=60)
if p.returncode != 0:
_link_log(url, "info", p.stderr.strip()[-300:])
raise ValueError(link_err(p.stderr))
try:
return json.loads(p.stdout)
except ValueError:
raise ValueError("video အချက်အလက် ဖတ်မရဘူး")
def fetch_link_info(url):
info = ytdlp_info(url)
if info.get("is_live"):
raise ValueError("live video လက်မခံဘူး")
dur = info.get("duration") or 0
if dur and dur > MAX_DURATION:
raise ValueError("၁၀ မိနစ်ထက် ရှည်တဲ့ ဗီဒီယို လက်မခံဘူး")
return {"title": (info.get("title") or "")[:120],
"duration": int(dur) if dur else None,
"thumbnail": info.get("thumbnail") or "",
"site": info.get("extractor_key") or info.get("extractor") or ""}
def download_link(url, d):
"""Download a pasted link into job dir d as input.<ext>. Returns ext."""
p = subprocess.run([str(YTDLP), "--no-playlist", "--no-warnings",
"--impersonate", "chrome", "-f", LINK_FMT,
"--merge-output-format", "mp4",
"--max-filesize", f"{MAX_MB}M",
"--socket-timeout", "20", "--retries", "3",
"-o", str(d / "input.%(ext)s"), url],
capture_output=True, text=True, timeout=600)
got = [f for f in d.glob("input.*")
if f.suffix.lower() in (".mp4", ".mov", ".mkv", ".webm", ".m4v")]
if p.returncode != 0 or not got:
for f in got: # drop partials from a failed download
f.unlink(missing_ok=True)
_link_log(url, "download", p.stderr.strip()[-300:])
raise ValueError(link_err(p.stderr))
return got[0].suffix.lower()
app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = MAX_MB * 1024 * 1024
jobs = {}
queue = []
lock = threading.Condition()
def jlog(job, msg):
"""Append a timestamped line to the job's live log file."""
line = f"[{time.strftime('%H:%M:%S')}] {msg}\n"
try:
with open(job["dir"] / "job.log", "a") as fh:
fh.write(line)
except OSError:
pass
def _is_quota(e):
"""True when a job died because a Gemini key ran out of quota."""
if isinstance(e, QuotaExhausted):
return True
s = str(e).lower()
return "resource_exhausted" in s or "quota" in s or "429" in s
def clean_err(e):
"""User-facing error text — never leak server paths."""
s = str(e)
if ("speech.platform.bing.com" in s or "WSServerHandshakeError" in s
or "Invalid response status" in s):
return ("အသံထုတ်တဲ့ service (Microsoft) ခေတ္တ ပျက်နေတယ် — "
"ပြန်ကောင်းအောင် ပြင်နေပါတယ်။ ခဏနေမှ ပြန်စမ်းပါ 🙏")
return s.replace(str(BASE), "[app]").replace(str(JOBS), "[jobs]")[:300]
def _client_ip():
# nginx on this box sets X-Real-IP. X-Forwarded-For alone is not
# reliable here: Waitress strips it unless the proxy is trusted,
# which left every logged IP as 127.0.0.1 until 2026-11-08.
real = request.headers.get("X-Real-IP", "").strip()
if real:
return real
fwd = request.headers.get("X-Forwarded-For", "")
return (fwd.split(",")[0].strip() or request.remote_addr or "?")
def _usage(event, **kw):
"""Append one usage event for the admin panel counter."""
try:
rec = {"ts": time.time(), "event": event, **kw}
with _usage_lock:
with open(USAGE, "a", encoding="utf-8") as fh:
fh.write(json.dumps(rec, ensure_ascii=False) + "\n")
except OSError:
pass # counting must never break a job
def _rate_ok(ip):
now = time.time()
with _ip_lock:
dq = _ip_hits.get(ip)
if dq is None:
dq = _ip_hits[ip] = deque()
while dq and now - dq[0] > 3600:
dq.popleft()
if len(dq) >= UPLOADS_PER_HOUR:
return False
dq.append(now)
if len(_ip_hits) > 5000: # don't let the table grow forever
_ip_hits.clear()
return True
# drop leftover job dirs from before a restart
for _old in JOBS.iterdir():
try:
if _old.is_dir() and time.time() - _old.stat().st_mtime > TTL:
shutil.rmtree(_old, ignore_errors=True)
except OSError:
pass
def sweep_old():
now = time.time()
for jid, job in list(jobs.items()):
if now - job["created"] > TTL and job["status"] in ("done", "failed"):
shutil.rmtree(job["dir"], ignore_errors=True)
jobs.pop(jid, None)
for sdir in STAGE.iterdir():
try:
if sdir.is_dir() and now - sdir.stat().st_mtime > STAGE_TTL:
shutil.rmtree(sdir, ignore_errors=True)
except OSError:
pass
def _parse_box(raw):
"""'x,y,w,h' blur box as 0..1 fractions of the frame -> tuple or None."""
try:
parts = [float(v) for v in str(raw or "").split(",")]
except ValueError:
return None
if len(parts) != 4:
return None
x, y, w, h = parts
if w < 0.02 or h < 0.02:
return None
x = min(max(x, 0.0), 0.98)
y = min(max(y, 0.0), 0.98)
return (x, y, min(w, 1.0 - x), min(h, 1.0 - y))
def _sub_opts(get):
"""Subtitle choices shared by the upload form and the JSON link API."""
subs = str(get("subs") or "") in ("1", "on", "true")
bg = str(get("sub_bg") or "box")
return subs, (bg if bg in ("box", "none") else "box")
def process_job(jid):
job = jobs[jid]
d = job["dir"]
try:
api_key = os.environ.get("GEMINI_API_KEY")
if not api_key:
raise RuntimeError("server config missing (GEMINI_API_KEY)")
direction, voice_name = job["direction"], VOICES[job["direction"]][job["voice"]]
lang_hint = "en" if direction == "en2mm" else "my"
engine = job.get("engine") or "nara"
set_engine(engine)
set_nara_voice(job.get("nara_voice"))
start_msg = (f"job {jid} started · {direction} · {job['voice']} voice "
f"· {engine} engine")
if engine == "nara" and job.get("nara_voice"):
start_msg += f" · {job['nara_voice']}"
jlog(job, start_msg)
if job.get("url"):
job.update(stage="download", pct=2)
jlog(job, f"downloading link: {job['url'][:100]}")
job["ext"] = download_link(job["url"], d)
mb = (d / ("input" + job["ext"])).stat().st_size / 1048576
jlog(job, f"downloaded ({mb:.1f} MB)")
src = d / ("input" + job["ext"])
total = probe_duration(src)
jlog(job, f"video duration: {total:.1f}s")
if total > MAX_DURATION:
raise ValueError("၁၀ မိနစ်ထက် ရှည်တဲ့ ဗီဒီယို လက်မခံဘူး")
if not has_video_stream(src):
raise ValueError("ဗီဒီယိုဖိုင် မဟုတ်ဘူး")
if not has_audio_stream(src):
raise ValueError("ဒီဗီဒီယိုမှာ အသံမပါဘူး — အသံပါတဲ့ ဗီဒီယိုမှ dub လို့ရမယ်")
if job.get("blur"):
# Hide the source video's own burned-in subtitles before
# anything else reads the picture; the box the user drew is
# stored as frame fractions, so scale it to real pixels here
job.update(stage="blur", pct=3)
vw, vh = probe_size(src)
fx, fy, fw, fh = job["blur"]
px = (int(fx * vw) // 2 * 2, int(fy * vh) // 2 * 2,
max(2, int(fw * vw) // 2 * 2), max(2, int(fh * vh) // 2 * 2))
jlog(job, f"blurring original subtitle area {px} of {vw}x{vh}…")
work = d / "work.mp4"
blur_region(src, work, px)
src = work
job.update(stage="audio", pct=4)
jlog(job, "extracting audio…")
wav = d / "audio.wav"
extract_audio(src, wav)
job.update(stage="transcribe", pct=8)
jlog(job, "transcribing with whisper…")
segs = transcribe(wav, lang_hint)
if not segs:
raise ValueError("အသံထဲမှာ စကားပြောသံ မတွေ့ဘူး")
utter = merge_segments(segs)
jlog(job, f"transcribed {len(segs)} segments → {len(utter)} utterances")
job.update(stage="translate", pct=45)
# HF Spaces has no OpenCode agent, so Luna is skipped there
# (DUBLAB_NO_LUNA=1) and Gemini translates directly.
if os.environ.get("DUBLAB_NO_LUNA"):
jlog(job, "translating with gemini…")
texts = translate_lines([u["text"] for u in utter], direction, api_key)
jlog(job, "translated with gemini")
else:
jlog(job, "translating with luna…")
try:
texts = luna_translate_lines([u["text"] for u in utter], direction)
jlog(job, "translated with luna")
except Exception as e: # noqa: BLE001 - Luna down/quota: Gemini next
jlog(job, f"Luna ခဏမရလို့ ({clean_err(e)}) — Gemini နဲ့ ဆက်ဘာသာပြန်တယ်")
try:
texts = translate_lines([u["text"] for u in utter], direction, api_key)
except Exception as e2: # noqa: BLE001 - Gemini down/quota: local fallback
jlog(job, f"Gemini ခဏမရလို့ ({clean_err(e2)}) — NLLB (local) နဲ့ ဆက်ဘာသာပြန်တယ်")
try:
from nllb_translate import translate_lines as nllb_translate_lines
texts = nllb_translate_lines([u["text"] for u in utter], direction)
except Exception:
# HF build ships no NLLB model: surface the Gemini
# error instead of masking it with a missing-model one.
raise e2
jlog(job, "translated with NLLB (local)")
for u, t in zip(utter, texts):
u["text"] = t or u["text"]
jlog(job, f"translated {len(texts)} lines")
job.update(stage="voice", pct=58)
jlog(job, f"synthesizing {len(utter)} utterances ({voice_name})…")
def tts_prog(i, n):
job.update(pct=58 + int(32 * i / max(n, 1)))
if i % 5 == 0 or i == n:
jlog(job, f"tts {i}/{n}")
dub_wav, warp, cues = build_dub_track(utter, voice_name, d, total, tts_prog)
slowed = [f for (_a, _b, f) in warp if f > 1.001]
if slowed:
jlog(job, f"slowed {len(slowed)} scene part(s) in the picture "
f"(max {max(slowed):.2f}x) so speech stays natural")
job.update(stage="mux", pct=93)
jlog(job, "muxing dubbed audio with video…")
ass_path = None
if job.get("subs"):
# Cues sit on the output timeline (build_dub_track placed
# them after stretch/cascade), so the ASS matches the final
# picture; the .srt twin is offered as a download
vw, vh = probe_size(src)
write_ass(cues, d / "subs.ass", vw, vh, job.get("sub_bg") or "box")
write_srt(cues, d / "subs.srt")
ass_path = d / "subs.ass"
jlog(job, f"burning in {len(cues)} subtitle lines "
f"(bg={job.get('sub_bg') or 'box'})…")
mux(src, dub_wav, d / "dubbed.mp4", warp, ass_path)
try:
cleanup_segments(d)
except OSError:
pass
job.update(status="done", stage="done", pct=100)
jlog(job, "done ✅")
_usage("done", job_id=jid, ip=job.get("ip", "?"),
secs=round(time.time() - job["created"]), dur=round(total))
quota_ok()
except Exception as e: # noqa: BLE001 - surfaced to UI
err = clean_err(e)
jlog(job, f"FAILED: {type(e).__name__}: {err}")
_usage("fail", job_id=jid, ip=job.get("ip", "?"), error=err[:80])
job.update(status="failed", error=err)
if _is_quota(e):
where = STAGE_LABEL.get(job["stage"], job["stage"])
quota_alert(f"{where} (job {jid})", err)
def worker():
while True:
with lock:
while not queue:
lock.wait()
jid = queue.pop(0)
jobs[jid]["status"] = "processing"
process_job(jid)
sweep_old()
threading.Thread(target=worker, daemon=True).start()
@app.get("/")
def index():
return render_template("index.html",
maintenance=MAINTENANCE,
maintenance_msg=MAINTENANCE_MSG)
@app.get("/health")
def health():
return {"ok": True}
@app.post("/api/jobs")
def create_job():
if MAINTENANCE:
return jsonify(error=MAINTENANCE_MSG), 503
if len(queue) >= MAX_QUEUE:
return jsonify(error="လူများနေတယ် — ခဏနေမှ ပြန်စမ်းပါ"), 429
if not _rate_ok(_client_ip()):
return jsonify(error="ခဏခဏ တင်နေတယ် — ၁ နာရီနေမှ ပြန်စမ်းပါ"), 429
f = request.files.get("video")
direction = request.form.get("direction", "en2mm")
voice = request.form.get("voice", "female")
engine = request.form.get("engine", "nara")
if engine not in ("edge", "gemini", "nara"):
engine = "nara"
nara_voice = valid_nara_voice(request.form.get("nara_voice"))
subs, sub_bg = _sub_opts(request.form.get)
blur = _parse_box(request.form.get("blur"))
if not f or not f.filename:
return jsonify(error="ဗီဒီယိုဖိုင် ရွေးပေးပါ"), 400
ext = os.path.splitext(f.filename)[1].lower()
if ext not in (".mp4", ".mov", ".mkv", ".webm", ".m4v"):
return jsonify(error="MP4/MOV ဖိုင်ပဲ လက်ခံတယ်"), 400
if direction not in VOICES or voice not in ("female", "male"):
return jsonify(error="ရွေးချယ်မှု မှားနေတယ်"), 400
jid = uuid.uuid4().hex[:12]
d = JOBS / jid
try:
d.mkdir(parents=True)
src = d / ("input" + ext)
f.save(src)
# fail fast on bad files — nothing reaches the worker queue
if probe_duration(src) > MAX_DURATION:
raise ValueError("၁၀ မိနစ်ထက် ရှည်တဲ့ ဗီဒီယို လက်မခံဘူး")
if not has_video_stream(src):
raise ValueError("ဗီဒီယိုဖိုင် မဟုတ်ဘူး")
except Exception as e: # noqa: BLE001
shutil.rmtree(d, ignore_errors=True)
if isinstance(e, ValueError):
return jsonify(error=clean_err(e)), 400
return jsonify(error="ဖိုင်စစ်ရင်း အမှားရှိတယ်"), 500
jobs[jid] = {"status": "queued", "stage": "queue", "pct": 0, "error": "",
"dir": d, "ext": ext, "direction": direction,
"voice": voice, "engine": engine, "nara_voice": nara_voice, "created": time.time(),
"subs": subs, "sub_bg": sub_bg, "blur": blur,
"ip": _client_ip()}
_usage("job", job_id=jid, ip=jobs[jid]["ip"], kind="file",
direction=direction, voice=voice, engine=engine)
fname = re.sub(r"[\r\n]", " ", f.filename or "?")[:80]
jlog(jobs[jid], f"uploaded {fname} ({src.stat().st_size / 1048576:.1f} MB)")
with lock:
queue.append(jid)
lock.notify()
return jsonify(job_id=jid)
@app.post("/api/jobs/check-url")
def check_url():
"""Look up a pasted link so the user sees title/duration before dubbing."""
if MAINTENANCE:
return jsonify(error=MAINTENANCE_MSG), 503
data = request.get_json(force=True, silent=True) or {}
try:
url = check_link(str(data.get("url") or ""))
info = fetch_link_info(url)
except ValueError as e:
return jsonify(error=str(e)), 400
except subprocess.TimeoutExpired:
return jsonify(error="site က တုံ့ပြန်တာ ကြာနေတယ် — ခဏနေ ပြန်စမ်းပါ"), 400
return jsonify(ok=True, url=url, **info)
def _extract_frames(sdir, sid, src, dur):
"""Five sample frames (720px wide) for the blur-box picker."""
frames = []
for i, frac in enumerate((0.08, 0.27, 0.46, 0.65, 0.84)):
out = sdir / f"frame{i}.jpg"
p = subprocess.run(
["ffmpeg", "-y", "-v", "error", "-ss", f"{dur * frac:.2f}",
"-i", str(src), "-frames:v", "1",
"-vf", "scale=720:-2", str(out)],
capture_output=True, text=True, timeout=60)
if p.returncode == 0 and out.exists():
frames.append(f"/api/stage/{sid}/frame{i}.jpg")
return frames
def _make_preview(sdir, src):
"""Small H.264 copy of the staged input for in-browser preview.
Safari's <video> cannot play some codecs a phone file may carry
(AV1 in an .mp4, seen 2026-11-07 — black preview, codec error
icon), while ffmpeg reads them fine. A 720px H.264/AAC faststart
copy plays everywhere, so the user can watch and scrub it while
marking the blur box.
"""
out = sdir / "preview.mp4"
p = subprocess.run(
["ffmpeg", "-y", "-v", "error", "-i", str(src),
"-vf", "scale='min(720,iw)':-2",
"-c:v", "libx264", "-preset", "veryfast", "-crf", "25",
"-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "96k",
"-movflags", "+faststart", str(out)],
capture_output=True, text=True, timeout=600)
if p.returncode != 0 or not out.exists():
raise ValueError("preview ထုတ်မရဘူး")
return out
@app.post("/api/jobs/link-frames")
def link_frames():
"""Download a link once and return sample frames for blur marking.
The download lands in a stage dir keyed by a session id; the client
draws the blur box over one of the frames, and from-url with the
same sid reuses this file instead of downloading again.
"""
if MAINTENANCE:
return jsonify(error=MAINTENANCE_MSG), 503
if not _rate_ok(_client_ip()):
return jsonify(error="ခဏခဏ တင်နေတယ် — ၁ နာရီနေမှ ပြန်စမ်းပါ"), 429
data = request.get_json(force=True, silent=True) or {}
try:
url = check_link(str(data.get("url") or ""))
except ValueError as e:
return jsonify(error=str(e)), 400
sid = uuid.uuid4().hex[:12]
sdir = STAGE / sid
sdir.mkdir(parents=True)
try:
ext = download_link(url, sdir)
src = sdir / ("input" + ext)
dur = probe_duration(src)
if dur > MAX_DURATION:
raise ValueError("၁၀ မိနစ်ထက် ရှည်တဲ့ ဗီဒီယို လက်မခံဘူး")
w, h = probe_size(src)
frames = _extract_frames(sdir, sid, src, dur)
if not frames:
raise ValueError("frame ထုတ်မရဘူး — file တိုက်ရိုက် တင်ပေးပါ")
except ValueError as e:
shutil.rmtree(sdir, ignore_errors=True)
return jsonify(error=str(e)), 400
except Exception: # noqa: BLE001
shutil.rmtree(sdir, ignore_errors=True)
return jsonify(error="frame ထုတ်မရဘူး — file တိုက်ရိုက် တင်ပေးပါ"), 400
return jsonify(sid=sid, w=w, h=h, duration=int(dur), frames=frames)
@app.get("/api/stage/<sid>/frame<int:i>.jpg")
def stage_frame(sid, i):
if not re.fullmatch(r"[0-9a-f]{12}", sid or "") or not 0 <= i < 5:
return jsonify(error="not found"), 404
p = STAGE / sid / f"frame{i}.jpg"
if not p.exists():
return jsonify(error="not found"), 404
return send_file(p)
@app.get("/api/stage/<sid>/preview.mp4")
def stage_preview(sid):
if not re.fullmatch(r"[0-9a-f]{12}", sid or ""):
return jsonify(error="not found"), 404
p = STAGE / sid / "preview.mp4"
if not p.exists():
return jsonify(error="not found"), 404
# conditional=True so Safari gets Range support and can scrub
return send_file(p, mimetype="video/mp4", conditional=True)
@app.post("/api/jobs/file-frames")
def file_frames():
"""Stage an uploaded file and build a playable preview from it.
Used when the browser cannot play the picked file itself (e.g. an
AV1 .mp4 in Safari): the file lands in a stage dir, we return
sample frames plus an H.264 preview the user can watch/scrub, and
from-file with the same sid reuses the upload as the job input —
no second upload.
"""
if MAINTENANCE:
return jsonify(error=MAINTENANCE_MSG), 503
if not _rate_ok(_client_ip()):
return jsonify(error="ခဏခဏ တင်နေတယ် — ၁ နာရီနေမှ ပြန်စမ်းပါ"), 429
f = request.files.get("video")
if not f or not f.filename:
return jsonify(error="ဗီဒီယိုဖိုင် ရွေးပေးပါ"), 400
ext = os.path.splitext(f.filename)[1].lower()
if ext not in (".mp4", ".mov", ".mkv", ".webm", ".m4v"):
return jsonify(error="MP4/MOV ဖိုင်ပဲ လက်ခံတယ်"), 400
sid = uuid.uuid4().hex[:12]
sdir = STAGE / sid
sdir.mkdir(parents=True)
try:
src = sdir / ("input" + ext)
f.save(src)
dur = probe_duration(src)
if dur > MAX_DURATION:
raise ValueError("၁၀ မိနစ်ထက် ရှည်တဲ့ ဗီဒီယို လက်မခံဘူး")
if not has_video_stream(src):
raise ValueError("ဗီဒီယိုဖိုင် မဟုတ်ဘူး")
w, h = probe_size(src)
frames = _extract_frames(sdir, sid, src, dur)
if not frames:
raise ValueError("frame ထုတ်မရဘူး")
_make_preview(sdir, src)
except ValueError as e:
shutil.rmtree(sdir, ignore_errors=True)
return jsonify(error=clean_err(e)), 400
except Exception: # noqa: BLE001
shutil.rmtree(sdir, ignore_errors=True)
return jsonify(error="preview ထုတ်မရဘူး"), 400
return jsonify(sid=sid, w=w, h=h, duration=int(dur), frames=frames,
preview=f"/api/stage/{sid}/preview.mp4")
@app.post("/api/jobs/from-file")
def create_job_from_file():
"""Create a job from a file already staged by file-frames."""
if MAINTENANCE:
return jsonify(error=MAINTENANCE_MSG), 503
if len(queue) >= MAX_QUEUE:
return jsonify(error="လူများနေတယ် — ခဏနေမှ ပြန်စမ်းပါ"), 429
if not _rate_ok(_client_ip()):
return jsonify(error="ခဏခဏ တင်နေတယ် — ၁ နာရီနေမှ ပြန်စမ်းပါ"), 429
data = request.get_json(force=True, silent=True) or {}
direction = str(data.get("direction") or "en2mm")
voice = str(data.get("voice") or "female")
engine = str(data.get("engine") or "nara")
if engine not in ("edge", "gemini", "nara"):
engine = "nara"
nara_voice = valid_nara_voice(data.get("nara_voice"))
subs, sub_bg = _sub_opts(data.get)
blur = _parse_box(data.get("blur"))
if direction not in VOICES or voice not in ("female", "male"):
return jsonify(error="ရွေးချယ်မှု မှားနေတယ်"), 400
sid = re.sub(r"[^0-9a-f]", "", str(data.get("sid") or ""))[:12]
staged = None
if sid:
sdir = STAGE / sid
if sdir.is_dir():
got = [f for f in sdir.glob("input.*")
if f.suffix.lower() in (".mp4", ".mov", ".mkv", ".webm", ".m4v")]
if got:
staged = got[0]
if staged is None:
return jsonify(error="preview session ကုန်သွားပြီ — file ပြန်ရွေးပေးပါ"), 400
jid = uuid.uuid4().hex[:12]
d = JOBS / jid
d.mkdir(parents=True)
ext = staged.suffix.lower()
shutil.move(str(staged), str(d / ("input" + ext)))
shutil.rmtree(STAGE / sid, ignore_errors=True)
jobs[jid] = {"status": "queued", "stage": "queue", "pct": 0, "error": "",
"dir": d, "ext": ext, "direction": direction,
"voice": voice, "engine": engine, "nara_voice": nara_voice, "created": time.time(),
"subs": subs, "sub_bg": sub_bg, "blur": blur,
"ip": _client_ip()}
_usage("job", job_id=jid, ip=jobs[jid]["ip"], kind="file",
direction=direction, voice=voice, engine=engine)
jlog(jobs[jid], "file job (staged preview session)")
with lock:
queue.append(jid)
lock.notify()
return jsonify(job_id=jid)
@app.post("/api/jobs/from-url")
def create_job_from_url():
if MAINTENANCE:
return jsonify(error=MAINTENANCE_MSG), 503
if len(queue) >= MAX_QUEUE:
return jsonify(error="လူများနေတယ် — ခဏနေမှ ပြန်စမ်းပါ"), 429
if not _rate_ok(_client_ip()):
return jsonify(error="ခဏခဏ တင်နေတယ် — ၁ နာရီနေမှ ပြန်စမ်းပါ"), 429
data = request.get_json(force=True, silent=True) or {}
direction = str(data.get("direction") or "en2mm")
voice = str(data.get("voice") or "female")
engine = str(data.get("engine") or "nara")
if engine not in ("edge", "gemini", "nara"):
engine = "nara"
nara_voice = valid_nara_voice(data.get("nara_voice"))
subs, sub_bg = _sub_opts(data.get)
blur = _parse_box(data.get("blur"))
if direction not in VOICES or voice not in ("female", "male"):
return jsonify(error="ရွေးချယ်မှု မှားနေတယ်"), 400
# A blur-marking session (link-frames) may already hold the download;
# if so the job reuses that file instead of downloading a second time
sid = re.sub(r"[^0-9a-f]", "", str(data.get("sid") or ""))[:12]
staged = None
if sid:
sdir = STAGE / sid
if sdir.is_dir():
got = [f for f in sdir.glob("input.*")
if f.suffix.lower() in (".mp4", ".mov", ".mkv", ".webm", ".m4v")]
if got:
staged = got[0]
try:
url = check_link(str(data.get("url") or ""))
if staged is None:
# fail fast: bad/private/too-long links never reach the queue
fetch_link_info(url)
except ValueError as e:
return jsonify(error=str(e)), 400
except subprocess.TimeoutExpired:
return jsonify(error="site က တုံ့ပြန်တာ ကြာနေတယ် — ခဏနေ ပြန်စမ်းပါ"), 400
jid = uuid.uuid4().hex[:12]
d = JOBS / jid
d.mkdir(parents=True)
jobs[jid] = {"status": "queued", "stage": "queue", "pct": 0, "error": "",
"dir": d, "ext": ".mp4", "direction": direction,
"voice": voice, "engine": engine, "nara_voice": nara_voice, "created": time.time(),
"subs": subs, "sub_bg": sub_bg, "blur": blur,
"ip": _client_ip()}
_usage("job", job_id=jid, ip=jobs[jid]["ip"], kind="link",
direction=direction, voice=voice, engine=engine)
if staged is not None:
ext = staged.suffix.lower()
shutil.move(str(staged), str(d / ("input" + ext)))
jobs[jid]["ext"] = ext
shutil.rmtree(STAGE / sid, ignore_errors=True)
else:
jobs[jid]["url"] = url
title = re.sub(r"[\r\n]", " ", str(data.get("title") or ""))[:80]
jlog(jobs[jid], f"link job: {title or url[:80]}")
with lock:
queue.append(jid)
lock.notify()
return jsonify(job_id=jid)
@app.get("/api/jobs/<jid>")
def job_status(jid):
job = jobs.get(jid)
if not job:
if re.fullmatch(r"[0-9a-f]{12}", jid or "") and (JOBS / jid / "dubbed.mp4").exists():
return jsonify(status="done", stage="done", pct=100,
label=STAGE_LABEL.get("done", "done"), error="",
has_srt=(JOBS / jid / "subs.srt").exists())
return jsonify(error="job မတွေ့ဘူး"), 404
return jsonify(status=job["status"], stage=job["stage"], pct=job["pct"],
label=STAGE_LABEL.get(job["stage"], job["stage"]),
error=job["error"], has_srt=bool(job.get("subs")))
@app.get("/api/jobs/<jid>/srt")
def job_srt(jid):
if not re.fullmatch(r"[0-9a-f]{12}", jid or ""):
return jsonify(error="not found"), 404
p = JOBS / jid / "subs.srt"
if not p.exists():
return jsonify(error="စာတန်း မရှိဘူး"), 404
return send_file(p, as_attachment=True, download_name="dublab-subs.srt")
@app.get("/api/jobs/<jid>/log")
def job_log(jid):
job = jobs.get(jid)
if not job:
return jsonify(error="job မတွေ့ဘူး"), 404
p = job["dir"] / "job.log"
return jsonify(log=p.read_text() if p.exists() else "")
@app.post("/api/reviews")
def add_review():
if not _rate_ok(_client_ip()):
return jsonify(error="ခဏနေမှ ပြန်စမ်းပါ"), 429
data = request.get_json(force=True, silent=True) or {}
try:
stars = int(data.get("stars", 0))
except (TypeError, ValueError):
stars = 0
msg = re.sub(r"\s+", " ", str(data.get("message") or "")).strip()[:500]
jid = re.sub(r"[^0-9a-f]", "", str(data.get("job_id") or ""))[:12]
if not (1 <= stars <= 5) and not msg:
return jsonify(error="star ဒါမှမဟုတ် စာတစ်ခုခု ပေးပါ"), 400
rec = {"id": uuid.uuid4().hex[:8], "ts": time.time(), "ip": _client_ip(),
"stars": stars if 1 <= stars <= 5 else 0,
"message": msg, "job_id": jid}
try:
with open(REVIEWS, "a", encoding="utf-8") as fh:
fh.write(json.dumps(rec, ensure_ascii=False) + "\n")
except OSError:
return jsonify(error="သိမ်းမရဘူး"), 500
return jsonify(ok=True)
@app.get("/admin/reviews")
def admin_reviews():
if not ADMIN_KEY or request.args.get("key") != ADMIN_KEY:
return "not found", 404
return render_template("admin.html")
def _read_reviews():
rows = []
if REVIEWS.exists():
for line in REVIEWS.read_text(encoding="utf-8").splitlines():
try:
rows.append(json.loads(line))
except ValueError:
pass
rows.sort(key=lambda r: r.get("ts", 0), reverse=True)
return rows
def _admin_ok():
return bool(ADMIN_KEY) and request.args.get("key") == ADMIN_KEY
def _read_usage():
rows = []
if USAGE.exists():
for line in USAGE.read_text(encoding="utf-8").splitlines():
try:
rows.append(json.loads(line))
except ValueError:
pass
rows.sort(key=lambda r: r.get("ts", 0))
return rows
@app.get("/admin/api/reviews")
def admin_api_reviews():
if not _admin_ok():
return jsonify(error="not found"), 404
rows = _read_reviews()
dist = {str(i): 0 for i in range(1, 6)}
total_stars = rated = 0
for r in rows:
s = r.get("stars", 0)
if 1 <= s <= 5:
dist[str(s)] += 1
total_stars += s
rated += 1
# usage counter: jobs ever created, outcomes, distinct user IPs
urows = _read_usage()
now = time.time()
made = [r for r in urows if r.get("event") == "job"]
users = {r.get("ip") for r in urows if r.get("ip")}
users |= {r.get("ip") for r in rows
if r.get("ip") and r.get("ip") != "127.0.0.1"}
users.discard(None)
usage = {
"jobs": len(made),
"done": sum(1 for r in urows if r.get("event") == "done"),
"fail": sum(1 for r in urows if r.get("event") == "fail"),
"users": len(users),
"today": sum(1 for r in made if now - r.get("ts", 0) < 86400),
"week": sum(1 for r in made if now - r.get("ts", 0) < 7 * 86400),
"recent": [{"ts": r.get("ts", 0), "ip": r.get("ip", "?"),
"kind": r.get("kind", ""), "event": r.get("event", ""),
"dur": r.get("dur"), "job_id": r.get("job_id", "")}
for r in urows[-60:]][::-1]}
return jsonify(reviews=rows[:300], stats={
"total": len(rows), "rated": rated,
"avg": round(total_stars / rated, 1) if rated else 0,
"dist": dist}, usage=usage)
@app.delete("/admin/api/reviews/<rid>")
def admin_del_review(rid):
if not _admin_ok():
return jsonify(error="not found"), 404
rid = re.sub(r"[^0-9a-f]", "", rid)[:8]
rows = _read_reviews()
kept = [r for r in rows if r.get("id") != rid]
if len(kept) == len(rows):
return jsonify(error="not found"), 404
try:
with open(REVIEWS, "w", encoding="utf-8") as fh:
for r in kept:
fh.write(json.dumps(r, ensure_ascii=False) + "\n")
except OSError:
return jsonify(error="write failed"), 500
return jsonify(ok=True)
@app.get("/r/<jid>")
def result(jid):
job = jobs.get(jid)
if job and job["status"] == "done":
f = job["dir"] / "dubbed.mp4"
elif re.fullmatch(r"[0-9a-f]{12}", jid or "") and (JOBS / jid / "dubbed.mp4").exists():
f = JOBS / jid / "dubbed.mp4"
else:
return jsonify(error="အဆင်သင့် မဖြစ်သေးဘူး"), 404
return send_file(f, as_attachment=True,
download_name="dublab-dubbed.mp4")
"""DubLab — short-video dubbing website. Upload video -> dubbed video."""
import json
import os
import re
import shutil
import subprocess
import sys
import threading
import time
import urllib.request
import uuid
from collections import deque
from pathlib import Path
from urllib.parse import urlparse
from flask import Flask, jsonify, render_template, request, send_file
from alerts import quota_alert, quota_ok
from dub import (blur_region, build_dub_track, cleanup_segments, configure_tts,
extract_audio, has_audio_stream, has_video_stream,
merge_segments, mux, probe_duration, probe_size, set_engine,
set_nara_voice, transcribe, valid_nara_voice, write_ass,
write_srt)
from gemini import translate_lines
from luna import translate_lines as luna_translate_lines
from gemini_tts import QuotaExhausted
# Gemini TTS key rotation (same 5-key pool style as the SRT site).
# GEMINI_TTS_API_KEYS is comma-separated; falls back to the single GEMINI_API_KEY.
_tts_keys = [k.strip() for k in
os.environ.get("GEMINI_TTS_API_KEYS",
os.environ.get("GEMINI_API_KEY", "")).split(",")
if k.strip()]
configure_tts(_tts_keys)
BASE = Path(__file__).parent
JOBS = BASE / "jobs"
JOBS.mkdir(exist_ok=True)
# Link-mode blur marking downloads the video here first so the user can
# pick a frame and draw the blur box; from-url then moves the file into
# the job instead of downloading twice. Swept after an hour.
STAGE = BASE / "stage"
STAGE.mkdir(exist_ok=True)
STAGE_TTL = 3600
MAX_DURATION = 600 # seconds (raised 5→10 min 2026-11-07, his order)
MAX_MB = 250
TTL = 24 * 3600 # keep results a day
# ---- abuse protection (public site) ----
MAX_QUEUE = 15 # max jobs waiting in line
UPLOADS_PER_HOUR = 10 # per client IP
_ip_hits = {}
_ip_lock = threading.Lock()
# ---- maintenance mode (TTS outage) ----
# Turn True to pause new dub jobs; the home page shows the maintenance image.
MAINTENANCE = False
MAINTENANCE_MSG = ("🔧 DubLab ခေတ္တ ရပ်ထားပါတယ် — Microsoft အသံထုတ် service "
"ပျက်နေလို့ ပြင်ဆင်နေပါတယ်။ ပြန်ကောင်းတာနဲ့ အကြောင်းကြားမယ် 🙏")
# ---- reviews ----
REVIEWS = BASE / "reviews.jsonl"
ADMIN_KEY = os.environ.get("DUBLAB_ADMIN_KEY", "")
# ---- usage counter (admin panel only) ----
# Job dirs are swept after a day, so without this append-only log the
# totals would be lost: one line per job created / done / failed.
USAGE = BASE / "usage.jsonl"
_usage_lock = threading.Lock()
VOICES = {
"en2mm": {"female": "my-MM-NilarNeural", "male": "my-MM-ThihaNeural"},
"mm2en": {"female": "en-US-AvaNeural", "male": "en-US-AndrewNeural"},
}
STAGE_LABEL = {
"queue": "တန်းစီနေတယ်",
"download": "ဗီဒီယို ဆွဲချနေတယ်",
"blur": "မူရင်း စာတန်း ဖုံးနေတယ်",
"audio": "အသံထုတ်ယူနေတယ်",
"transcribe": "စကားပြောမှတ်နေတယ်",
"translate": "ဘာသာပြန်နေတယ်",
"voice": "အသံဖော်နေတယ်",
"mux": "ဗီဒီယိုပေါင်းနေတယ်",
"done": "ပြီးပြီ",
}
# ---- video link support (yt-dlp) ----
# Instead of uploading a file, users can paste a link from a social site.
# yt-dlp lives in the same venv as this app.
YTDLP = Path(sys.executable).parent / "yt-dlp"
LINK_HOSTS = ("youtube.com", "youtu.be", "tiktok.com", "instagram.com",
"facebook.com", "fb.watch", "xiaohongshu.com", "xhslink.com",
"rednote.com", "twitter.com", "x.com", "vimeo.com",
"dailymotion.com", "bilibili.com")
# best mp4 up to 720p; the /b fallback covers sites with combined formats only
LINK_FMT = "bv*[height<=720]+ba/b[height<=720]/b"
def _host_ok(host):