app.py

"""
صافي — Hugging Face Space (ZeroGPU)
فصل الموسيقى عن الكلام بموديل Mel-Band RoFormer (Kim) على GPU مجاني.

• واجهة Gradio (بيستخدمها الناس مباشرة)
• وكمان API تلقائي (واجهة Cloudflare تقدر تناديه)

ملاحظات ZeroGPU المتّبعة هنا:
- import spaces + @spaces.GPU على دالة الفصل بس
- الموديل بيتحمّل مرة واحدة في الـ global (read-only) — مش جوّه الدالة
- مفيش حالة عامة قابلة للتعديل، ومسارات الإخراج فريدة (tempfile) لأن
  الطلبات بتشتغل بالتوازي افتراضياً
- مفيش torch.compile (ممنوع على ZeroGPU)
"""
import os
import sys
import glob
import tempfile
import subprocess

import gradio as gr
import spaces

# ---------- رقعة لـ bug معروف في gradio_client (additionalProperties=bool) ----------
# بيكسر بناء الـ API schema برسالة: argument of type 'bool' is not iterable
import gradio_client.utils as _gcu

_orig_json = _gcu._json_schema_to_python_type
_orig_get_type = _gcu.get_type


def _safe_get_type(schema):
    if not isinstance(schema, dict):
        return "Any"
    return _orig_get_type(schema)


def _safe_json(schema, defs=None):
    if isinstance(schema, bool):
        return "Any"
    return _orig_json(schema, defs)


_gcu.get_type = _safe_get_type
_gcu._json_schema_to_python_type = _safe_json

# ---------- تحميل الموديل مرة واحدة (read-only global) ----------
# نخلّي حزمة melband تنزّل الموديل + الإعدادات بنفسها (بصيغة متوافقة مع safe_load)
import glob as _glob

MODEL_SLUG = "melband-roformer-kim-vocals"
MODELS_DIR = os.path.join(tempfile.gettempdir(), "safi_models")
os.makedirs(MODELS_DIR, exist_ok=True)

# ننزّل الموديل بأمر الحزمة المباشر (مثبت إنه على الـ PATH وقت التشغيل)
_model_dir = os.path.join(MODELS_DIR, MODEL_SLUG)
if not os.path.isdir(_model_dir) or not _glob.glob(os.path.join(_model_dir, "*.ckpt")):
    subprocess.run(
        ["melband-roformer-download",
         "--model", MODEL_SLUG, "--output-dir", MODELS_DIR],
        check=True,
    )

_cfgs = _glob.glob(os.path.join(_model_dir, "*.yaml"))
_ckpts = _glob.glob(os.path.join(_model_dir, "*.ckpt"))
if not _cfgs or not _ckpts:
    raise RuntimeError(f"تحميل الموديل فشل، مفيش ملفات في {_model_dir}")
CONFIG_PATH = _cfgs[0]
CKPT_PATH = _ckpts[0]


def _to_wav(src, dst):
    """تحويل أي صيغة لـ WAV ستيريو 44.1kHz."""
    subprocess.run(
        ["ffmpeg", "-y", "-i", src, "-vn", "-acodec", "pcm_s16le",
         "-ar", "44100", "-ac", "2", dst],
        check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )


@spaces.GPU(duration=60)
def _separate_gpu(in_dir, out_dir):
    """الجزء اللي محتاج GPU: تشغيل الموديل على جزء واحد. (ZeroGPU)"""
    subprocess.run(
        ["melband-roformer-infer",
         "--config_path", CONFIG_PATH,
         "--model_path", CKPT_PATH,
         "--input_folder", in_dir,
         "--store_dir", out_dir],
        check=True,
    )


# ---------- التقسيم لأجزاء + التجميع مع cross-fade ----------
import numpy as _np
import soundfile as _sf

CHUNK_SEC = 20.0      # طول كل جزء (ثواني) — كل جزء = استدعاء GPU قصير
OVERLAP_SEC = 1.0     # تداخل بين الأجزاء (عشان مفيش قطع مسموع)


def _separate_in_chunks(wav_path, work, progress=None):
    """
    بيقسّم الملف لأجزاء، يفصل كل جزء في استدعاء GPU قصير،
    ويجمّعهم بنظام cross-fade. بيرجّع (vocals_path, music_path).
    """
    audio, sr = _sf.read(wav_path, always_2d=True)   # (عينات, قنوات)
    total = audio.shape[0]
    ch = audio.shape[1]
    chunk = int(CHUNK_SEC * sr)
    ov = int(OVERLAP_SEC * sr)
    step = chunk - ov

    # حدود الأجزاء
    spans = []
    s = 0
    while s < total:
        e = min(s + chunk, total)
        spans.append((s, e))
        if e >= total:
            break
        s += step

    voc = _np.zeros((total, ch), dtype=_np.float64)
    mus = _np.zeros((total, ch), dtype=_np.float64)
    wsum = _np.zeros((total, 1), dtype=_np.float64)

    for i, (s, e) in enumerate(spans):
        if progress is not None:
            progress((i, len(spans)), desc=f"بيفصل الجزء {i+1}/{len(spans)}")
        cdir = os.path.join(work, f"c{i}")
        cin = os.path.join(cdir, "in")
        cout = os.path.join(cdir, "out")
        os.makedirs(cin, exist_ok=True)
        os.makedirs(cout, exist_ok=True)
        _sf.write(os.path.join(cin, "track.wav"), audio[s:e], sr)

        _separate_gpu(cin, cout)   # استدعاء GPU قصير لكل جزء

        v = glob.glob(os.path.join(cout, "*vocals*.wav"))
        m = glob.glob(os.path.join(cout, "*instrumental*.wav"))
        if not v:
            raise gr.Error(f"الفصل فشل في الجزء {i+1}")
        vd, _ = _sf.read(v[0], always_2d=True)
        md = _sf.read(m[0], always_2d=True)[0] if m else _np.zeros_like(vd)

        n = e - s
        vd = vd[:n]; md = md[:n]
        w = _np.ones((n, 1))
        if ov > 0:
            ramp = _np.linspace(0, 1, min(ov, n)).reshape(-1, 1)
            if s > 0:
                w[:len(ramp)] *= ramp
            if e < total:
                w[-len(ramp):] *= ramp[::-1]
        voc[s:e] += vd * w
        mus[s:e] += md * w
        wsum[s:e] += w

    wsum[wsum < 1e-8] = 1.0
    voc = (voc / wsum).astype(_np.float32)
    mus = (mus / wsum).astype(_np.float32)

    vpath = os.path.join(work, "vocals.wav")
    mpath = os.path.join(work, "music.wav")
    _sf.write(vpath, voc, sr)
    _sf.write(mpath, mus, sr)
    return vpath, mpath


def _download_youtube(url, out_path):
    """ينزّل أفضل صوت من لينك يوتيوب بـ yt-dlp إلى out_path (wav)."""
    # ملاحظة: للتجربة الشخصية فقط — تنزيل يوتيوب يخالف شروطه.
    # ننادي yt-dlp كموديول بايثون (أأمن من الاعتماد على وجوده كأمر في PATH)
    base = out_path.rsplit(".", 1)[0]
    subprocess.run(
        [sys.executable, "-m", "yt_dlp",
         "-x", "--audio-format", "wav", "--audio-quality", "0",
         "-o", base + ".%(ext)s", url],
        check=True,
    )


def separate_link(url):
    """تجريبي: بياخد لينك يوتيوب، ينزّل الصوت، يفصله، ويرجّع (الكلام, الموسيقى)."""
    if not url or not url.strip():
        return None, None
    work = tempfile.mkdtemp()
    dl_dir = os.path.join(work, "dl")     # التنزيل هنا
    os.makedirs(dl_dir, exist_ok=True)

    raw = os.path.join(dl_dir, "yt.wav")
    _download_youtube(url.strip(), raw)
    got = glob.glob(os.path.join(dl_dir, "*.wav"))
    if not got:
        raise gr.Error("مش قادر أنزّل الصوت من اللينك ده.")
    wav = os.path.join(work, "track.wav")
    _to_wav(got[0], wav)
    return _separate_in_chunks(wav, work)


def separate(audio_file, progress=gr.Progress()):
    """بتاخد ملف، تحوّله، تفصله جزء جزء، وترجّع (الكلام, الموسيقى)."""
    if audio_file is None:
        return None, None
    work = tempfile.mkdtemp()                 # فريد لكل طلب
    wav = os.path.join(work, "track.wav")
    _to_wav(audio_file, wav)                   # CPU
    return _separate_in_chunks(wav, work, progress)   # GPU جزء جزء


# ---------- واجهة Gradio (بهوية صافي) ----------
CSS = """
.gradio-container{max-width:760px!important;margin:auto}
#safi-title{font-weight:900;font-size:30px;text-align:center;color:#3dd7b8}
#safi-sub{text-align:center;color:#8a97a3;margin-bottom:8px}
footer{visibility:hidden}
"""

with gr.Blocks(css=CSS, title="صافي — إزالة الموسيقى", theme=gr.themes.Soft()) as demo:
    gr.Markdown("# ● صافي", elem_id="safi-title")
    gr.Markdown("اسمع الكلام، سيبك من الموسيقى — فصل بموديل Mel-Band RoFormer", elem_id="safi-sub")
    with gr.Tab("ارفع ملف"):
        inp = gr.Audio(label="ارفع ملف صوت", type="filepath")
        btn = gr.Button("افصل الموسيقى", variant="primary")
    with gr.Tab("لينك يوتيوب (تجريبي)"):
        gr.Markdown(
            "⚠️ للتجربة الشخصية فقط — تنزيل يوتيوب يخالف شروطه ومش مناسب للنشر العام.",
            elem_id="safi-sub",
        )
        link = gr.Textbox(label="لينك يوتيوب", placeholder="https://www.youtube.com/watch?v=...")
        btn_link = gr.Button("نزّل وافصل", variant="primary")
    with gr.Row():
        out_vocals = gr.Audio(label="الكلام (صافي)")
        out_music = gr.Audio(label="الموسيقى")
    btn.click(fn=separate, inputs=inp, outputs=[out_vocals, out_music], api_name="separate")
    btn_link.click(fn=separate_link, inputs=link, outputs=[out_vocals, out_music], api_name="separate_link")
    gr.Markdown(
        "الترخيص MIT — تجاري. أنت مسؤول عن حقوق المحتوى اللي ترفعه.",
        elem_id="safi-sub",
    )

# api_name بيخلّي الدالة متاحة كـ API لواجهة Cloudflare
demo.queue()  # مهم لـ ZeroGPU
if __name__ == "__main__":
    # show_api=False بيتجنّب bug توليد الـ schema في gradio_client
    demo.launch()