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"] = 512 * 1024 * 1024  # S146: 512MB — supports up to 200MB files

BASE = Path(__file__).parent  # S149: moved before JOBS_FILE — was NameError
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)

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

# ── Imp-1: Job persistence — survives space restart/sleep ─────────────────────
JOBS_FILE = BASE / "tilawa_jobs.json"

def _save_jobs():
    """Persist active jobs to disk (excludes locks and file handles)."""
    import json
    _SKIP = {"_lock"}
    try:
        with JOBS_LOCK:
            snapshot = {}
            for jid, j in JOBS.items():
                if j.get("status") in ("done", "error"):
                    continue  # completed jobs don't need persistence
                row = {}
                for k, v in j.items():
                    if k in _SKIP:
                        continue
                    if isinstance(v, set):
                        row[k] = list(v)
                    else:
                        row[k] = v
                snapshot[jid] = row
        with open(JOBS_FILE, "w") as f:
            json.dump(snapshot, f)
    except Exception as e:
        print(f"[persist] save failed: {e}")

def _load_jobs():
    """Reload jobs from disk on startup. Mark running/queued as error."""
    import json
    if not JOBS_FILE.exists():
        return
    try:
        with open(JOBS_FILE) as f:
            snapshot = json.load(f)
        loaded = 0
        with JOBS_LOCK:
            for jid, j in snapshot.items():
                # recreate threading.Lock
                j["_lock"] = threading.Lock()
                # restore set
                if "received_set" in j:
                    j["received_set"] = set(j["received_set"])
                # jobs that were mid-run can't resume — mark as error
                if j.get("status") in ("running", "queued", "merging", "uploading"):
                    j["status"] = "error"
                    j["label"]  = "فشل: أُعيد تشغيل الخادم — أرسل الملف مجدداً"
                    j["error"]  = "Server restarted — please resubmit"
                JOBS[jid] = j
                loaded += 1
        print(f"[persist] loaded {loaded} job(s) from disk")
    except Exception as e:
        print(f"[persist] load failed: {e}")

_load_jobs()

ENGINE_SCRIPTS = {
    "v11.0": BASE / "engine_tajalli_v1.py",
    "v11.1": BASE / "true_engine_itiqan_v2_fixed.py",
    "v11.2": BASE / "engine_isteidad_v21.py",
    "v10.0": BASE / "engine_tajalli_v1.py",
    "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

_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 _timeout_stuck_jobs():
    """Mark jobs stuck in running/queued/merging > 30min as error."""
    cutoff = time.time() - 1800
    with JOBS_LOCK:
        for job in JOBS.values():
            if job.get("status") in ("running", "queued", "merging"):
                ts = job.get("_started_at", time.time())
                if ts < cutoff:
                    job["status"] = "error"
                    job["error"]  = "Job timed out after 30 minutes"
                    job["label"]  = "فشل: انتهت مهلة المعالجة (30 دقيقة)"

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

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

def _keepalive():
    import urllib.request
    time.sleep(120)
    # S146: use SPACE_HOST so every space pings itself, not just tilawa-server
    host = os.environ.get("SPACE_HOST", "carm5333-tilawa-server.hf.space")
    url  = os.environ.get("SPACE_URL", f"https://{host}/ping")
    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()
    ram_gb   = _available_ram_gb()
    try:
        disk_gb = shutil.disk_usage(TMP).free / (1024**3)
    except Exception:
        disk_gb = -1
    with JOBS_LOCK:
        queued = sum(1 for j in JOBS.values() if j.get("status") == "queued")
    return jsonify({"status": "ok", "engines": engines,
                    "refs": len(refs), "chunk_size": CHUNK_SIZE,
                    "running": _count_running(), "queued": queued,
                    "ram_gb": round(ram_gb, 2),
                    "disk_free_gb": round(disk_gb, 2)})  # Imp-C

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

@app.route("/wake")  # Imp-3: Flutter pre-warm endpoint
def wake():
    engines = {k: v.exists() for k, v in ENGINE_SCRIPTS.items()}
    return jsonify({"ok": True, "t": time.time(),
                    "engines": engines, "running": _count_running()})

@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
    content_length = request.content_length or 0
    if content_length > 0 and not _check_disk_free(content_length * 2):
        return jsonify({"error": "server disk full — try later"}), 503  # Imp-D
    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["_started_at"] = time.time()  # Imp-A
        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()
            cmd = ["python3", str(script),
                   "-i", job["in_path"], "-o", job["out_path"],
                   "--iterations", "3"]
            for rf in ref_files[:3]: cmd += ["--ref", str(rf)]
            _ENGINE_TIMEOUT = 25 * 60  # Imp-B: 25min hard limit
            proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
                                    stderr=subprocess.STDOUT, text=True)
            engine_log = []
            for line in proc.stdout:
                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 "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 50.0 <= s <= 100.0:
                                job["score"] = s; job["progress"] = 95
                                job["label"] = "حساب النتيجة..."
                        except Exception: pass
            try:
                proc.wait(timeout=_ENGINE_TIMEOUT)  # Imp-B
            except subprocess.TimeoutExpired:
                proc.kill(); proc.wait()
                job["engine_log"] = engine_log
                job["status"] = "error"
                job["label"]  = "فشل: تجاوز وقت المحرك (25 دقيقة)"
                job["error"]  = "Engine timed out after 25 minutes"
                _save_jobs()
                return
            job["engine_log"] = engine_log
            _out = Path(job["out_path"])
            if _out.exists() and _out.stat().st_size > 0:
                success = True
            else:
                job["engine_rc"] = proc.returncode
        except Exception as exc:
            job["engine_error"] = str(exc)
    if not success:
        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}"
            _save_jobs()  # Imp-1: persist error state
            return
    with JOBS_LOCK:
        job["status"] = "done"; job["progress"] = 100; job["label"] = "اكتملت ✓"
        if not job.get("score"): job["score"] = 90
        job.pop("engine_log", None)  # Imp-4: trim log after done — save RAM
    _add_history(job)
    _save_jobs()  # Imp-1: persist
    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
             ["engine","filename","score","lufs","rms","crest","lra","timestamp"]
             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"]
    if job.get("status") == "error":
        log = job.get("engine_log", [])
        resp["engine_log"] = log[-5:] if log else []  # Imp-E: last 5 lines
        resp["engine_rc"]  = job.get("engine_rc")
    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})

# S68: auto-download missing engine files from GitHub on startup
def _bootstrap_engines():
    import urllib.request
    RAW = "https://raw.githubusercontent.com/hammer24678-star/tilawa-enhancer-/master/assets/engines"
    engines = [
        "engine_tajalli_v1.py",
        "true_engine_itiqan_v2_fixed.py",
        "engine_isteidad_v21.py",
        "engine_v90.py",
        "engine_v85.py",
        "engine_v80.py",
        "engine_v70.py",
    ]
    for e in engines:
        p = BASE / e
        if not p.exists() or p.stat().st_size < 1000:
            try:
                urllib.request.urlretrieve(f"{RAW}/{e}", str(p))
                print(f"[bootstrap] downloaded {e}")
            except Exception as ex:
                print(f"[bootstrap] failed {e}: {ex}")

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

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