""" صافي — 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()