Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
| """Web server: `python -m uvicorn app.server:app --port 8000` then open http://localhost:8000""" | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import shutil | |
| import threading | |
| import time | |
| import traceback | |
| import uuid | |
| from concurrent.futures import ThreadPoolExecutor | |
| from pathlib import Path | |
| from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile | |
| from fastapi.responses import FileResponse, StreamingResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from . import engines, feedback, languages, pipeline, safety | |
| DATA = engines.DATA | |
| JOBS_DIR = DATA / "jobs" | |
| JOBS_DIR.mkdir(parents=True, exist_ok=True) | |
| STATIC = Path(__file__).parent / "static" | |
| app = FastAPI(title="SwitchLingo") | |
| feedback.start() | |
| pool = ThreadPoolExecutor(max_workers=1) # one job at a time keeps CPU usage sane | |
| JOBS: dict[str, dict] = {} | |
| # Public-use limits (override with environment variables) | |
| TERMS_VERSION = "2026-11-07" | |
| MAX_MINUTES = float(os.environ.get("OXT_MAX_MINUTES", "10")) | |
| MAX_MB = int(os.environ.get("OXT_MAX_MB", "300")) | |
| MAX_PER_IP = int(os.environ.get("OXT_MAX_PER_IP", "2")) # active jobs per visitor | |
| MAX_QUEUE = int(os.environ.get("OXT_MAX_QUEUE", "20")) | |
| KEEP_HOURS = float(os.environ.get("OXT_KEEP_HOURS", "24")) # outputs are deleted after this | |
| def _cleanup_loop() -> None: | |
| while True: | |
| cutoff = time.time() - KEEP_HOURS * 3600 | |
| for d in JOBS_DIR.iterdir(): | |
| try: | |
| if d.is_dir() and d.stat().st_mtime < cutoff: | |
| shutil.rmtree(d, ignore_errors=True) | |
| JOBS.pop(d.name, None) | |
| except OSError: | |
| pass | |
| time.sleep(1800) | |
| threading.Thread(target=_cleanup_loop, daemon=True).start() | |
| def _client_ip(request: Request) -> str: | |
| fwd = request.headers.get("x-forwarded-for", "") | |
| return fwd.split(",")[0].strip() if fwd else (request.client.host if request.client else "?") | |
| def _save(job: dict) -> None: | |
| disk = {k: v for k, v in job.items() if k != "ip"} # visitor IPs stay in memory only | |
| (JOBS_DIR / job["id"] / "job.json").write_text(json.dumps(disk, ensure_ascii=False)) | |
| def _work(job_id: str, kwargs: dict) -> None: | |
| job = JOBS[job_id] | |
| job["status"] = "running" | |
| def progress(msg: str, pct: float) -> None: | |
| job["step"], job["progress"] = msg, round(pct, 3) | |
| try: | |
| job["result"] = pipeline.run(JOBS_DIR / job_id, progress=progress, **kwargs) | |
| job["status"] = "done" | |
| except Exception as e: # noqa: BLE001 | |
| job["status"], job["error"] = "error", str(e) | |
| traceback.print_exc() | |
| job["finished"] = time.time() | |
| _save(job) | |
| def get_languages(): | |
| return {"languages": languages.public_list(), "popular": languages.POPULAR_US, | |
| "limits": {"max_minutes": MAX_MINUTES, "max_mb": MAX_MB}, | |
| "translators": engines.available_translators(), | |
| "voices": engines.available_voices(), "lipsync": engines.lipsync_available()} | |
| async def create_job( | |
| request: Request, | |
| target: str = Form(...), | |
| url: str = Form(""), | |
| source: str = Form(""), | |
| engine: str = Form("madlad"), | |
| outputs: str = Form("text,subs,dub"), | |
| quality: str = Form("fast"), | |
| gender: str = Form("Female"), | |
| voice_engine: str = Form("edge"), | |
| clone: str = Form("1"), | |
| lipsync: str = Form("0"), | |
| agree: str = Form(""), | |
| voice_consent: str = Form(""), | |
| age: str = Form(""), | |
| clean: str = Form("0"), | |
| file: UploadFile | None = File(None), | |
| ): | |
| if age not in ("teen", "adult"): | |
| raise HTTPException(400, "Please tell us your age group. SwitchLingo is for ages 13 and up.") | |
| if agree != "1": | |
| raise HTTPException(400, "Please confirm you have the right to use this video and accept the Terms.") | |
| if age == "teen": # Teen mode (13-17): clean words, standard AI voice, no cloning/lip-sync/premium | |
| clone, lipsync, clean, voice_engine = "0", "0", "1", "edge" | |
| if clone == "1" and voice_consent != "1": | |
| clone = "0" # never clone a voice without the user's permission statement | |
| if target not in languages.LANGS: | |
| raise HTTPException(400, "Unknown target language") | |
| if engine not in engines.available_translators(): | |
| raise HTTPException(400, f"Translation engine '{engine}' is not set up") | |
| if voice_engine not in engines.available_voices(): | |
| raise HTTPException(400, f"Voice engine '{voice_engine}' is not set up") | |
| if not url.strip() and not (file and file.filename): | |
| raise HTTPException(400, "Give a link or upload a file") | |
| ip = _client_ip(request) | |
| active = [j for j in JOBS.values() if j["status"] in ("queued", "running")] | |
| if len(active) >= MAX_QUEUE: | |
| raise HTTPException(429, "Busy right now - please try again in a few minutes") | |
| if sum(1 for j in active if j.get("ip") == ip) >= MAX_PER_IP: | |
| raise HTTPException(429, "You already have videos in progress - wait for them to finish") | |
| job_id = uuid.uuid4().hex[:12] | |
| jdir = JOBS_DIR / job_id | |
| jdir.mkdir(parents=True) | |
| src_path = None | |
| if file and file.filename: | |
| ext = Path(file.filename).suffix.lower() or ".bin" | |
| src_path = jdir / f"upload{ext}" | |
| size = 0 | |
| with open(src_path, "wb") as f: | |
| while chunk := await file.read(1 << 20): | |
| size += len(chunk) | |
| if size > MAX_MB * 1024 * 1024: | |
| f.close() | |
| shutil.rmtree(jdir, ignore_errors=True) | |
| raise HTTPException(413, f"File is over {MAX_MB} MB") | |
| f.write(chunk) | |
| job = {"id": job_id, "status": "queued", "step": "Waiting in line", "progress": 0.0, | |
| "created": time.time(), "input": file.filename if file and file.filename else url.strip(), | |
| "target": target, "ip": ip, | |
| "consent": {"terms": TERMS_VERSION, "rights": True, "voice": clone == "1", "age": age, | |
| "time": time.time()}} | |
| JOBS[job_id] = job | |
| _save(job) | |
| kwargs = dict(source_path=src_path, url=url.strip() or None, target=target, | |
| source_lang=source or None, engine=engine, | |
| outputs=tuple(o for o in outputs.split(",") if o), quality=quality, | |
| gender=gender, voice_engine=voice_engine, max_minutes=MAX_MINUTES, | |
| clone=clone == "1", lipsync=lipsync == "1" and engines.lipsync_available(), | |
| clean=clean == "1") | |
| pool.submit(_work, job_id, kwargs) | |
| return {"id": job_id} | |
| def get_job(job_id: str): | |
| job = JOBS.get(job_id) | |
| if not job: | |
| p = JOBS_DIR / job_id / "job.json" | |
| if not p.exists(): | |
| raise HTTPException(404, "Job not found") | |
| job = json.loads(p.read_text()) | |
| return {k: v for k, v in job.items() if k != "ip"} # never expose visitor IPs | |
| def list_jobs(): | |
| jobs = [] | |
| for p in sorted(JOBS_DIR.glob("*/job.json"), key=lambda x: x.stat().st_mtime, reverse=True)[:30]: | |
| j = JOBS.get(p.parent.name) or json.loads(p.read_text()) | |
| jobs.append({k: j.get(k) for k in ("id", "status", "input", "target", "created")}) | |
| if os.environ.get("OXT_PUBLIC", "1") == "1": # public site: don't list other people's jobs | |
| jobs = [] | |
| return {"jobs": jobs} | |
| def get_file(job_id: str, name: str, download: bool = False): | |
| p = (JOBS_DIR / job_id / name).resolve() | |
| if not p.is_file() or JOBS_DIR.resolve() not in p.parents: | |
| raise HTTPException(404) | |
| return FileResponse(p, filename=name if download else None) | |
| def language_request(name: str = Form(...), region: str = Form(""), notes: str = Form(""), | |
| contact: str = Form("")): | |
| rec = {"name": name.strip(), "region": region.strip(), "notes": notes.strip(), | |
| "contact": contact.strip(), "time": time.time()} | |
| with open(DATA / "language_requests.jsonl", "a", encoding="utf-8") as f: | |
| f.write(json.dumps(rec, ensure_ascii=False) + "\n") | |
| return {"ok": True} | |
| def post_feedback(request: Request, job_id: str = Form(...), seg: int = Form(...), rating: int = Form(0), | |
| correction: str = Form(""), token: str = Form("")): | |
| job = JOBS.get(job_id) | |
| if not job: | |
| p = JOBS_DIR / job_id / "job.json" | |
| job = json.loads(p.read_text()) if p.exists() else None | |
| if not job or job.get("status") != "done": | |
| raise HTTPException(404, "Job not found") | |
| r = job["result"] | |
| if not 0 <= seg < len(r["segments"]) or rating not in (-1, 0, 1): | |
| raise HTTPException(400, "Bad feedback") | |
| s = r["segments"][seg] | |
| correction = correction.strip()[:1000] | |
| if not rating and not correction: | |
| raise HTTPException(400, "Nothing to save") | |
| feedback.add({"job": job_id, "seg": seg, "src": r["source_lang"], "tgt": r["target_lang"], | |
| "engine": r.get("engine"), "original": s["original"], "machine": s["translated"], | |
| "rating": rating, "correction": correction, | |
| "voter": feedback.voter_id(_client_ip(request)), | |
| "trusted": bool(feedback.ADMIN_TOKEN) and token == feedback.ADMIN_TOKEN}) | |
| return {"ok": True} | |
| TEACH_PROMPTS = [ | |
| "Good morning, how are you?", "I love you, Mom.", "Where can I find the bus stop?", | |
| "Please wait here, I'll be right back.", "How old are you?", "Thank you, see you tomorrow.", | |
| "The doctor will see you now.", "Do you need help?", | |
| "Where is the nearest hospital?", "How much does this cost?", "Thank you for helping my family.", | |
| "What time does the store close?", "I don't understand, please speak slowly.", | |
| "My child is sick and needs a doctor.", "Can you help me fill out this form?", | |
| "The food is delicious.", "I am looking for work.", "Please call me back tomorrow.", | |
| "Grandma is coming to visit next week.", "Where do I pay my bill?", | |
| "I need a translator for my appointment.", "Happy New Year to everyone!", | |
| "Be careful, the road is slippery.", | |
| ] | |
| # Dialects and languages the main models barely cover. Taught sentences in these are kept for training | |
| # (not mixed into the parent language's translation memory). Anyone can also type any other language. | |
| TEACH_DIALECTS = { | |
| "lo-isan": "Isan (Northeast Thai / Lao Isan)", "hmn": "Hmong (White / Hmoob Dawb)", | |
| "hmn-green": "Hmong (Green / Hmoob Leeg)", "mien": "Iu Mien", "kar": "Karen (S'gaw)", "cnh": "Chin (Hakha)", | |
| "nan": "Hokkien / Taiwanese", "tw-hak": "Hakka", "teo": "Teochew", "tsn-tai": "Taishanese (Toisan)", | |
| "chk": "Chuukese", "mh": "Marshallese", "sm-ws": "Samoan", "to": "Tongan", "ilo": "Ilocano", | |
| "ceb": "Cebuano (Bisaya)", "pam": "Kapampangan", "mam": "Mam (Maya)", "quc": "K'iche' (Maya)", | |
| "nv": "Navajo (Diné)", "chr": "Cherokee", "ht": "Haitian Creole", "pap": "Papiamento", | |
| "ti": "Tigrinya", "om": "Oromo", "din": "Dinka", "nus": "Nuer", "rw": "Kinyarwanda", "ff": "Fula (Pulaar)", | |
| "ar-lev": "Arabic (Levantine)", "ar-gulf": "Arabic (Gulf)", "ar-eg": "Arabic (Egyptian)", | |
| "ar-mag": "Arabic (Moroccan / Darija)", "es-mx": "Spanish (Mexico)", "es-cari": "Spanish (Caribbean)", | |
| "pt-br": "Portuguese (Brazil)", "fr-af": "French (West Africa)", "en-aave": "English (AAVE)", | |
| "la": "Latin", "grc": "Ancient Greek", "sa": "Sanskrit", | |
| } | |
| def _teach_langs() -> dict[str, str]: | |
| base = {r["code"]: r["name"] for r in languages.public_list()} | |
| return {**base, **TEACH_DIALECTS} | |
| def teach_languages(): | |
| base = {r["code"]: r["name"] for r in languages.public_list()} | |
| return {"languages": base, "dialects": TEACH_DIALECTS, "popular": languages.POPULAR_US, | |
| "prompts": TEACH_PROMPTS} | |
| async def teach(request: Request, lang: str = Form(...), en: str = Form(...), text: str = Form(...), | |
| token: str = Form(""), agree: str = Form(""), age: str = Form(""), other: str = Form(""), | |
| audio: UploadFile | None = File(None)): | |
| if agree != "1": | |
| raise HTTPException(400, "Please accept how we use your sentence and recording.") | |
| if age not in ("teen", "adult"): | |
| raise HTTPException(400, "Please tell us your age group (13+).") | |
| if age == "teen" and audio and audio.filename: | |
| raise HTTPException(400, "Voice recordings are for ages 18+. You can still add the sentence as text.") | |
| allowed = _teach_langs() | |
| other = " ".join(other.split())[:60] | |
| if lang == "other": | |
| if len(other) < 2: | |
| raise HTTPException(400, "Type the name of your language or dialect") | |
| elif lang not in allowed: | |
| raise HTTPException(400, "Unknown language") | |
| en, text = en.strip()[:500], text.strip()[:500] | |
| if not en or not text: | |
| raise HTTPException(400, "Both sentences are needed") | |
| data, ext = None, "webm" | |
| if audio and audio.filename: | |
| data = await audio.read(3 * 1024 * 1024 + 1) | |
| if len(data) > 3 * 1024 * 1024: | |
| raise HTTPException(413, "Recording is too long (max ~1 minute)") | |
| ext = (Path(audio.filename).suffix.lstrip(".") or "webm")[:5] | |
| base = lang if lang in languages.LANGS else ("other" if lang == "other" else lang) | |
| feedback.add_pair(base, en, text, feedback.voter_id(_client_ip(request)), | |
| bool(feedback.ADMIN_TOKEN) and token == feedback.ADMIN_TOKEN, | |
| dialect=(other if lang == "other" else | |
| TEACH_DIALECTS.get(lang, "") if lang not in languages.LANGS else ""), | |
| audio=data, audio_ext=ext) | |
| return {"ok": True} | |
| def report_kinds(): | |
| return {"kinds": safety.KINDS} | |
| def post_report(kind: str = Form(...), link: str = Form(""), details: str = Form(""), name: str = Form(""), | |
| contact: str = Form(""), sworn: str = Form("")): | |
| if kind not in safety.KINDS: | |
| raise HTTPException(400, "Choose what's wrong") | |
| if not contact.strip(): | |
| raise HTTPException(400, "We need a way to reach you (email or phone)") | |
| if kind == "copyright" and sworn != "1": | |
| raise HTTPException(400, "Copyright notices need the good-faith statement (DMCA).") | |
| import re | |
| m = re.search(r"(?:job=|/api/files/|^)([0-9a-f]{12})\b", link.strip()) | |
| job = m.group(1) if m else "" | |
| rec = safety.add_report(JOBS_DIR, {"kind": kind, "job": job, "link": link.strip()[:500], | |
| "details": details.strip()[:3000], "name": name.strip()[:200], | |
| "contact": contact.strip()[:200], "sworn": sworn == "1"}) | |
| if rec.get("auto_removed"): | |
| JOBS.pop(job, None) | |
| return {"ok": True, "id": rec["id"], "removed": bool(rec.get("auto_removed"))} | |
| def _admin(token: str) -> None: | |
| if not feedback.ADMIN_TOKEN or token != feedback.ADMIN_TOKEN: | |
| raise HTTPException(403, "Admin token required") | |
| def admin_reports(token: str = ""): | |
| _admin(token) | |
| return {"reports": safety.reports()[::-1]} | |
| def admin_remove(token: str = Form(""), job_id: str = Form(...), reason: str = Form("owner")): | |
| _admin(token) | |
| JOBS.pop(job_id, None) | |
| return {"removed": safety.quarantine(JOBS_DIR, job_id, reason)} | |
| def admin_restore(token: str = Form(""), job_id: str = Form(...)): | |
| _admin(token) | |
| return {"restored": safety.restore(JOBS_DIR, job_id)} | |
| def feedback_stats(): | |
| return feedback.stats() | |
| def feedback_export(token: str = ""): | |
| if not feedback.ADMIN_TOKEN or token != feedback.ADMIN_TOKEN: | |
| raise HTTPException(403, "Admin token required") | |
| return StreamingResponse(feedback.export_lines(), media_type="application/x-ndjson", | |
| headers={"Content-Disposition": "attachment; filename=feedback.jsonl"}) | |
| def share_target_fallback(): | |
| # Normally the service worker handles shares; this only runs before it's installed. | |
| from fastapi.responses import RedirectResponse | |
| return RedirectResponse("/", status_code=303) | |
| app.mount("/", StaticFiles(directory=STATIC, html=True), name="static") | |