Blame for app/server.py

"""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)


@app.get("/api/languages")
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()}


@app.post("/api/jobs")
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}


@app.get("/api/jobs/{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


@app.get("/api/jobs")
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}


@app.get("/api/files/{job_id}/{name}")
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)


@app.post("/api/language-requests")
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}


@app.post("/api/feedback")
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}


@app.get("/api/teach/languages")
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}


@app.post("/api/teach")
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}


@app.get("/api/report-kinds")
def report_kinds():
    return {"kinds": safety.KINDS}


@app.post("/api/reports")
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")


@app.get("/api/admin/reports")
def admin_reports(token: str = ""):
    _admin(token)
    return {"reports": safety.reports()[::-1]}


@app.post("/api/admin/remove")
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)}


@app.post("/api/admin/restore")
def admin_restore(token: str = Form(""), job_id: str = Form(...)):
    _admin(token)
    return {"restored": safety.restore(JOBS_DIR, job_id)}


@app.get("/api/feedback/stats")
def feedback_stats():
    return feedback.stats()


@app.get("/api/feedback/export")
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"})


@app.post("/share-target")
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")