app.py

"""
tilawa-server app.py — v7 (S62 Production)

Key changes:
  - ENGINE_SCRIPTS uses real filenames deployed on HF Space
  - SEMAPHORE=3 (matches HF free tier 16GB RAM)
  - Self-ping URL configurable via SPACE_URL env var
  - All v5 thread-safety, disk/RAM guards, janitor retained
"""
import gc, os, re as _re, shutil, subprocess, tempfile
import threading, time, uuid
from pathlib import Path
from flask import Flask, Response, jsonify, request, stream_with_context
from flask_cors import CORS

app = Flask(__name__)
CORS(app)
app.config["MAX_CONTENT_LENGTH"] = 6 * 1024 * 1024

JOBS         = {}
HISTORY      = []
JOBS_LOCK    = threading.Lock()
HISTORY_LOCK = threading.Lock()
SEMAPHORE    = threading.Semaphore(3)

TMP        = Path(tempfile.gettempdir())
UPLOAD_DIR = TMP / "tilawa_uploads"
CHUNK_DIR  = TMP / "tilawa_chunks"
OUTPUT_DIR = TMP / "tilawa_outputs"
for _d in [UPLOAD_DIR, CHUNK_DIR, OUTPUT_DIR]:
    _d.mkdir(exist_ok=True)

BASE = Path(__file__).parent

ENGINE_SCRIPTS = {
    "v11.0": BASE / "engine_safaa_v5.py",       # S191: v5 continuous DF3 blend + S190 WPE guard       # S175  # S149
    "v11.1": BASE / "engine_itiqan_v6_server.py",   # S200: server-optimised الإتقان
    "v11.2": BASE / "engine_isteidad_server.py",  # S214: server-hardened isteidad (إحياء v4 removed)
    "v10.0": BASE / "engine_safaa_v3_fixed.py",  # S149
    "v9.0":  BASE / "engine_v90.py",
    "v8.9":  BASE / "engine_v89.py",
    "v8.7":  BASE / "engine_v87.py",
    "v8.5":  BASE / "engine_v85.py",
    "v8.4":  BASE / "engine_v84.py",
    "v8.0":  BASE / "engine_v80.py",
    "v7.0":  BASE / "engine_v70.py",
}
DEFAULT_ENGINE     = "v11.0"
REF_DIR            = BASE / "reference_audio"
CHUNK_SIZE         = 4 * 1024 * 1024
MAX_DOWNLOAD_CHUNK = 32 * 1024 * 1024
ENGINE_TIMEOUT     = 600  # S185: 300s was killing ~7min recordings mid-pass

_REF_CACHE      = None
_REF_CACHE_TS   = 0.0
_REF_CACHE_LOCK = threading.Lock()
_REF_CACHE_TTL  = 3600

def _get_ref_files():
    global _REF_CACHE, _REF_CACHE_TS
    now = time.time()
    with _REF_CACHE_LOCK:
        if _REF_CACHE is None or (now - _REF_CACHE_TS) > _REF_CACHE_TTL:
            _REF_CACHE    = list(REF_DIR.glob("*.mp3")) if REF_DIR.exists() else []
            _REF_CACHE_TS = now
        return list(_REF_CACHE)

def _sanitize_filename(name):
    name = Path(name).name
    name = _re.sub(r"[^\w\-. ]", "_", name)
    return (name[:200] or "audio")

def _available_ram_gb():
    try:
        mem = {}
        with open("/proc/meminfo") as fh:
            for line in fh:
                if line.startswith(("MemAvailable:", "MemTotal:")):
                    k, v = line.split(":", 1)
                    mem[k.strip()] = int(v.split()[0])
        avail = mem.get("MemAvailable", 0)
        if avail > 0: return avail / (1024 * 1024)
        total = mem.get("MemTotal", 0)
        if total > 0: return total / (1024 * 1024)
    except Exception: pass
    return 999.0

def _check_disk_free(required_bytes):
    try: return shutil.disk_usage(TMP).free >= required_bytes
    except Exception: return True

def _get_queue_position(job_id):
    with JOBS_LOCK:
        queued = [jid for jid, j in JOBS.items() if j.get("status") == "queued"]
    try: return queued.index(job_id) + 1
    except ValueError: return 0

def _count_running():
    with JOBS_LOCK:
        return sum(1 for j in JOBS.values() if j.get("status") == "running")

def _prune_jobs():
    with JOBS_LOCK:
        if len(JOBS) <= 200: return
        removable = [jid for jid, j in list(JOBS.items())
                     if j.get("status") in ("done", "error")]
        for jid in removable[:-100]: JOBS.pop(jid, None)

def _cleanup_old_outputs():
    cutoff = time.time() - 7200
    try:
        for f in OUTPUT_DIR.iterdir():
            try:
                if f.stat().st_mtime < cutoff: f.unlink()
            except Exception: pass
    except Exception: pass

def _cleanup_stale_chunks():
    cutoff = time.time() - 14400
    try:
        for d in CHUNK_DIR.iterdir():
            if d.is_dir():
                try:
                    if d.stat().st_mtime < cutoff:
                        shutil.rmtree(d, ignore_errors=True)
                except Exception: pass
    except Exception: pass

def _janitor():
    while True:
        time.sleep(1800)
        for fn in (_cleanup_old_outputs, _prune_jobs, _cleanup_stale_chunks):
            try: fn()
            except Exception: pass

threading.Thread(target=_janitor, daemon=True).start()

def _keepalive():
    import urllib.request
    time.sleep(120)
    url = os.environ.get("SPACE_URL", "http://127.0.0.1:7860/ping")  # S162-SV-B2
    while True:
        try: urllib.request.urlopen(url, timeout=15)
        except Exception: pass
        time.sleep(240)

threading.Thread(target=_keepalive, daemon=True).start()

@app.route("/")
def health():
    engines = {k: v.exists() for k, v in ENGINE_SCRIPTS.items()}
    refs    = _get_ref_files()
    return jsonify({"status": "ok", "engines": engines,
                    "refs": len(refs), "chunk_size": CHUNK_SIZE,
                    "running": _count_running()})

@app.route("/ping")
def ping():
    return jsonify({"ok": True, "t": time.time()})

@app.route("/queue")
def queue_status():
    with JOBS_LOCK:
        running = sum(1 for j in JOBS.values() if j.get("status") == "running")
        queued  = sum(1 for j in JOBS.values() if j.get("status") == "queued")
    return jsonify({"running": running, "queued": queued,
                    "capacity": 3, "available": max(0, 3 - running)})

@app.route("/ready")
def ready():
    return jsonify({"ready": True,
                    "engines": {k: v.exists() for k, v in ENGINE_SCRIPTS.items()}})

@app.route("/upload", methods=["POST"])
def upload():
    if "file" not in request.files:
        return jsonify({"error": "no file"}), 400
    f      = request.files["file"]
    engine = request.form.get("engine", DEFAULT_ENGINE)
    if engine not in ENGINE_SCRIPTS: engine = DEFAULT_ENGINE
    job_id  = str(uuid.uuid4())[:16]
    suffix  = Path(_sanitize_filename(f.filename or "audio.mp3")).suffix or ".mp3"
    in_path = UPLOAD_DIR / f"{job_id}_input{suffix}"
    f.save(str(in_path))
    _init_job(job_id, engine, str(in_path), original_name=f.filename or "audio")
    threading.Thread(target=_run_engine_queued, args=(job_id,), daemon=True).start()
    return jsonify({"job_id": job_id})

@app.route("/upload_start", methods=["POST"])
def upload_start():
    data       = request.get_json(silent=True) or {}
    filename   = _sanitize_filename(data.get("filename", "audio.mp3"))
    total_size = int(data.get("total_size", 0))
    required   = max(total_size * 2, 10 * 1024 * 1024)
    if not _check_disk_free(required):
        return jsonify({"error": "server disk full — try later"}), 503
    total_chunks = max(1, -(-total_size // CHUNK_SIZE))
    job_id = str(uuid.uuid4())[:16]
    suffix = Path(filename).suffix or ".mp3"
    job = {
        "status": "uploading", "progress": 0, "label": "جارٍ الرفع...",
        "engine": DEFAULT_ENGINE, "fcm_token": data.get("fcm_token", ""),
        "filename": f"enhanced_{job_id}_1425h.mp3",
        "in_path":  str(UPLOAD_DIR / f"{job_id}_input{suffix}"),
        "out_path": str(OUTPUT_DIR / f"enhanced_{job_id}_1425h.mp3"),
        "score": None, "lufs": None, "rms": None, "crest": None, "lra": None,
        "timestamp": time.strftime("%Y-%m-%d %H:%M"),
        "suffix": suffix, "total_chunks": total_chunks,
        "received_chunks": 0, "received_set": set(), "total_size": total_size,
        "_lock": threading.Lock(),
    }
    with JOBS_LOCK: JOBS[job_id] = job
    (CHUNK_DIR / job_id).mkdir(exist_ok=True)
    return jsonify({"job_id": job_id, "chunk_size": CHUNK_SIZE,
                    "total_chunks": total_chunks})

@app.route("/upload_chunk", methods=["POST"])
def upload_chunk():
    job_id = request.form.get("job_id")
    try: index = int(request.form.get("index", 0))
    except (ValueError, TypeError):
        return jsonify({"error": "invalid index"}), 400
    with JOBS_LOCK: job = JOBS.get(job_id)
    if not job_id or job is None:
        return jsonify({"error": "invalid job_id"}), 400
    if "chunk" not in request.files:
        return jsonify({"error": "no chunk"}), 400
    total = job["total_chunks"]
    if not (0 <= index < total):
        return jsonify({"error": f"index out of range [0, {total})"}), 400
    chunk_path = CHUNK_DIR / job_id / f"chunk_{index:04d}"
    with job["_lock"]:
        if index not in job["received_set"]:
            request.files["chunk"].save(str(chunk_path))
            job["received_set"].add(index)
            job["received_chunks"] = len(job["received_set"])
        received = job["received_chunks"]
        missing  = [i for i in range(total) if i not in job["received_set"]]
    job["progress"] = int((received / total) * 30)
    job["label"]    = f"رفع {received}/{total}..."
    return jsonify({"received": received, "total": total,
                    "ok": True, "missing": missing})

@app.route("/upload_finalize", methods=["POST"])
def upload_finalize():
    data   = request.get_json(silent=True) or {}
    job_id = data.get("job_id")
    engine = data.get("engine", DEFAULT_ENGINE)
    if engine not in ENGINE_SCRIPTS: engine = DEFAULT_ENGINE
    with JOBS_LOCK: job = JOBS.get(job_id)
    if not job_id or job is None:
        return jsonify({"error": "invalid job_id"}), 400
    with job["_lock"]:
        total   = job["total_chunks"]
        missing = [i for i in range(total) if i not in job["received_set"]]
    if missing:
        return jsonify({"error": f"missing {len(missing)} chunk(s): {missing[:5]}",
                        "missing": missing}), 400
    job["engine"] = engine
    job["status"] = "merging"
    job["label"]  = "دمج الأجزاء..."
    job["progress"] = 32
    threading.Thread(target=_merge_and_run, args=(job_id,), daemon=True).start()
    return jsonify({"job_id": job_id, "status": "merging"})

def _init_job(job_id, engine, in_path, original_name="audio"):
    out_name = f"enhanced_{job_id}_1425h.mp3"
    with JOBS_LOCK:
        JOBS[job_id] = {
            "status": "queued", "progress": 5, "label": "في الطابور...",
            "engine": engine, "filename": out_name, "in_path": in_path,
            "out_path": str(OUTPUT_DIR / out_name),
            "score": None, "lufs": None, "rms": None, "crest": None, "lra": None,
            "timestamp": time.strftime("%Y-%m-%d %H:%M"),
            "_lock": threading.Lock(),
        }

def _merge_and_run(job_id):
    with JOBS_LOCK: job = JOBS.get(job_id)
    if not job: return
    chunk_dir = CHUNK_DIR / job_id
    in_path   = Path(job["in_path"])
    try:
        chunks = sorted(chunk_dir.glob("chunk_*"),
                        key=lambda p: int(p.stem.split("_")[1]))
        if not chunks: raise RuntimeError("No chunks found")
        with open(in_path, "wb") as out_f:
            for cp in chunks:
                out_f.write(cp.read_bytes()); cp.unlink()
        try: chunk_dir.rmdir()
        except Exception: pass
        job["progress"] = 35
        job["label"]    = f"تم الدمج — {in_path.stat().st_size // 1024 // 1024}MB"
        job["status"]   = "queued"
        _run_engine_queued(job_id)
    except Exception as e:
        job["status"] = "error"; job["error"] = str(e)
        job["label"]  = f"خطأ في الدمج: {e}"

def _run_engine_queued(job_id):
    with JOBS_LOCK: job = JOBS.get(job_id)
    if not job: return
    job["status"] = "queued"
    job["label"]  = "في الطابور — انتظر قليلاً..."
    SEMAPHORE.acquire()
    try: _run_engine(job_id)
    finally: SEMAPHORE.release(); gc.collect()

def _run_engine(job_id):
    with JOBS_LOCK: job = JOBS.get(job_id)
    if not job: return
    ram_gb = _available_ram_gb()
    if ram_gb < 0.5:
        with JOBS_LOCK:
            job["status"] = "error"
            job["label"]  = "خطأ: ذاكرة غير كافية — أعد المحاولة لاحقاً"
        return
    script = ENGINE_SCRIPTS.get(job["engine"])
    with JOBS_LOCK:
        job["status"]   = "running"
        job["progress"] = max(job.get("progress", 0), 35)
        job["label"]    = f"تشغيل المحرك {job['engine']}..."
    success = False
    if script and script.exists():
        try:
            ref_files = _get_ref_files()
            # S165: safaa has no --iterations / --ref — skip them
            _is_safaa = "safaa" in str(script).lower()
            # S173 Bug2: safaa uses soundfile → cannot write MP3
            # → write WAV first, then convert to MP3 with ffmpeg
            _o_path = (
                job["out_path"].replace(".mp3", "_safaa_tmp.wav")
                if _is_safaa else job["out_path"]
            )
            # S174-SV-C1: safaa uses positional args (input output), not -i/-o flags
            if _is_safaa:
                cmd = ["python3", str(script), job["in_path"], _o_path,
                       "--tier",     job.get("tier", "TIER_UNKNOWN"),
                       "--mujawwad", str(job.get("mujawwad_conf", 0.0))]  # S175
            else:
                cmd = ["python3", str(script),
                       "-i", job["in_path"], "-o", _o_path,
                       "--iterations", "3"]
                for rf in ref_files[:3]: cmd += ["--ref", str(rf)]
                if "itiqan" in str(script).lower():
                    cmd.append("--aggressive")  # S185: always aggressive, all spaces
            proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                                    stderr=subprocess.STDOUT, text=True)
            engine_log = []
            # S176: communicate() drains stdout AND enforces timeout atomically.
            # The old `for line in proc.stdout:` blocked until process exit,
            # making proc.wait(timeout=…) unreachable — silent hangs ate semaphores.
            _timed_out = False
            try:
                _out, _ = proc.communicate(timeout=ENGINE_TIMEOUT)
            except subprocess.TimeoutExpired:
                proc.kill()
                try:
                    _out, _ = proc.communicate(timeout=5)  # S189: drain buffered output instead of discarding it
                except Exception:
                    _out = ""
                _timed_out = True
                job["engine_error"] = (
                    f"engine killed after {ENGINE_TIMEOUT}s — possible hang")
            job["engine_rc"] = proc.returncode  # S185: always recorded
            for line in _out.splitlines():
                line = line.strip()
                engine_log.append(line)
                if len(engine_log) > 60: engine_log.pop(0)
                if   "Pass 1" in line or "[٧]" in line or "[١]" in line:
                    job["progress"] = 45; job["label"] = "Pass 1 — تحليل الطيف..."
                elif "Pass 2" in line or "[٨]" in line or "[٢]" in line:
                    job["progress"] = 60; job["label"] = "Pass 2 — ضبط LUFS..."
                elif "Pass 3" in line or "[٩]" in line or "[٣]" in line:
                    job["progress"] = 75; job["label"] = "Pass 3 — تصحيح..."
                elif "Pass 4" in line or "[٤]" in line:
                    job["progress"] = 88; job["label"] = "Pass 4 — تشفير MP3..."
                elif "[S1]" in line:
                    job["progress"] = 40; job["label"] = "تحليل الصدى (RT60)..."  # S187
                elif "[S2" in line:
                    job["progress"] = 46; job["label"] = "إزالة الرنين الأرضي..."  # S187
                elif "[S3-WPE]" in line:
                    job["progress"] = 54; job["label"] = "إزالة الريفيرب (WPE)..."  # S187
                elif "[S4-DF3]" in line or "[S4-ADF]" in line:
                    job["progress"] = 66; job["label"] = "تنقية الضجيج (DF3)..."  # S187
                elif "[S5-JALAA]" in line:
                    job["progress"] = 80; job["label"] = "تصحيح الديناميكية (JALAA)..."  # S187
                elif "[S6-tailNR]" in line:
                    job["progress"] = 90; job["label"] = "إزالة الضجيج المتبقي..."  # S187
                elif "[S7" in line:
                    job["progress"] = 94; job["label"] = "ضبط مخارج الحروف..."  # S187
                elif "LUFS=" in line:
                    for part in line.split():
                        try:
                            if   "LUFS="  in part: job["lufs"]  = part.split("=")[1]
                            elif "RMS="   in part: job["rms"]   = part.split("=")[1]
                            elif "Crest=" in part: job["crest"] = part.split("=")[1]
                            elif "LRA="   in part: job["lra"]   = part.split("=")[1]
                        except Exception: pass
                if "/100" in line:
                    m = _re.search(r"(\d{2,3}\.?\d*)\s*/\s*100", line)
                    if m:
                        try:
                            s = float(m.group(1))
                            if 0.0 <= s <= 100.0:  # S185: was 50-100, hid real bad scores
                                job["score"] = s; job["progress"] = 95
                                job["label"] = "حساب النتيجة..."
                        except Exception: pass
            # S176: proc timeout now handled inside communicate() above
            job["engine_log"] = engine_log
            # S173 Bug2: engine wrote to _o_path (may be .wav for safaa)
            _out = Path(_o_path)
            _clean = (not _timed_out) and proc.returncode == 0  # S185
            if _clean and _out.exists() and _out.stat().st_size > 0:
                if _is_safaa and _o_path != job["out_path"]:
                    subprocess.run(
                        ["ffmpeg", "-y", "-i", str(_out),
                         "-b:a", "320k", "-ar", "48000", job["out_path"]],
                        check=True, capture_output=True, timeout=120)
                    try: _out.unlink()
                    except Exception: pass
                success = True
            elif not _clean:
                # S185: killed or non-zero rc — don't trust a partial file,
                # force the real ffmpeg fallback instead of faking success
                job["engine_error"] = job.get("engine_error") or f"engine exited rc={proc.returncode}"
        except Exception as exc:
            job["engine_error"] = str(exc)
    if not success:
        job["used_fallback"] = True
        try:
            job["label"] = "معالجة أساسية (fallback)..."
            job["progress"] = 50
            subprocess.run(["ffmpeg", "-y", "-i", job["in_path"],
                            "-af", "loudnorm=I=-6:TP=-1:LRA=4",
                            "-ar", "48000", "-b:a", "320k", job["out_path"]],
                           check=True, capture_output=True, timeout=600)
            success = True; job["score"] = 75
        except Exception as e:
            with JOBS_LOCK:
                job["status"] = "error"; job["error"] = str(e)
                job["label"]  = f"فشل: {e}"
            return
    with JOBS_LOCK:
        job["status"] = "done"; job["progress"] = 100; job["label"] = "اكتملت ✓"
        if job.get("score") is None and not job.get("used_fallback"):
            job["score"] = 90  # S185: only on a clean, non-fallback run
    _add_history(job)
    try: Path(job["in_path"]).unlink(missing_ok=True)
    except Exception: pass
    _prune_jobs()

def _add_history(job):
    entry = {k: job[k] for k in
             ["status","error","engine","filename","score","lufs","rms","crest","lra",
              "timestamp","used_fallback","engine_rc","engine_error","engine_log",
              "report_json"]  # S173 Bug4
             if k in job}
    with HISTORY_LOCK:
        HISTORY.insert(0, entry)
        if len(HISTORY) > 50: HISTORY.pop()

@app.route("/status/<job_id>")
def status(job_id):
    with JOBS_LOCK: job = JOBS.get(job_id)
    if not job: return jsonify({"error": "not found"}), 404
    resp = {k: job[k] for k in
            ["status","progress","label","score","lufs","rms","crest","lra","filename"]
            if k in job}
    resp["queue_position"] = _get_queue_position(job_id) if job.get("status") == "queued" else 0
    if "error" in job: resp["error"] = job["error"]
    # S167: always expose engine diagnostics (incl. fallback runs)
    resp["engine_log"]    = job.get("engine_log", [])
    resp["engine_rc"]     = job.get("engine_rc")
    resp["engine_error"]  = job.get("engine_error")
    resp["used_fallback"] = job.get("used_fallback", False)
    return jsonify(resp)

@app.route("/download/<job_id>")
def download(job_id):
    with JOBS_LOCK: job = JOBS.get(job_id)
    if not job or job["status"] != "done":
        return jsonify({"error": "not ready"}), 404
    path = Path(job["out_path"])
    if not path.exists():
        return jsonify({"error": "file missing — may have expired"}), 404
    file_size = path.stat().st_size
    def generate():
        with open(path, "rb") as f:
            while True:
                chunk = f.read(1024 * 1024)
                if not chunk: break
                yield chunk
    return Response(stream_with_context(generate()), headers={
        "Content-Disposition": f'attachment; filename="{job["filename"]}"',
        "Content-Type": "audio/mpeg", "Content-Length": str(file_size)})

@app.route("/download_chunk/<job_id>")
def download_chunk(job_id):
    with JOBS_LOCK: job = JOBS.get(job_id)
    if not job or job["status"] != "done":
        return jsonify({"error": "not ready"}), 404
    path = Path(job["out_path"])
    if not path.exists(): return jsonify({"error": "file missing"}), 404
    file_size = path.stat().st_size
    try:
        offset = int(request.args.get("offset", 0))
        size   = int(request.args.get("size", CHUNK_SIZE))
    except (ValueError, TypeError):
        return jsonify({"error": "invalid offset or size"}), 400
    offset = max(0, min(offset, file_size))
    size   = max(0, min(size, MAX_DOWNLOAD_CHUNK, file_size - offset))
    with open(path, "rb") as f:
        f.seek(offset); data = f.read(size)
    return Response(data, headers={"Content-Type": "audio/mpeg",
                                   "Content-Length": str(len(data)),
                                   "X-File-Size": str(file_size),
                                   "X-Offset": str(offset)})

@app.route("/history")
def history():
    with HISTORY_LOCK: snapshot = list(HISTORY)
    return jsonify({"jobs": snapshot})

if __name__ == "__main__":
    port = int(os.environ.get("PORT", 7860))
    app.run(host="0.0.0.0", port=port, threaded=True)