Update app.py

app.py

@@ -1,107 +1,217 @@
-import os
-import sys
-from datetime import datetime
-
-# 🦾 تأمين الحزم والمفسرات الأساسية صامتاً في ذاكرة السيرفر
-try:
-    import transformers
-    import torchaudio
-except ImportError:
-    os.system(f"{sys.executable} -m pip install --upgrade pip -q")
-    os.system(f"{sys.executable} -m pip install transformers torchaudio accelerate gradio -q")
-
 import gradio as gr
+import json
+from datetime import datetime
+import yaml
+import time
+import re
+import os
+import os.path as op
 import torch
-import torchaudio
+import soundfile as sf
 import numpy as np
-from transformers import AutoModelForCausalLM, AutoTokenizer
+import tempfile
+
+from download import download_model
 
-MODEL_ID = "Novix/SongGenerationtwo"
-APP_DIR = os.path.dirname(os.path.abspath(__file__))
+# 🎯 خطوة السيادة الأولى: توجيه المحرك لتحميل أوزانكِ الخاصة بدلاً من المستودع العام
+APP_DIR = op.dirname(op.abspath(__file__))
+MODEL_ID = "Novix/SongGenerationtwo"  # مستودع أوزانكِ السيادية
 
-print("⏳ [Novix Core] جاري شحن المحرك السيادي المستقل...")
+print(f"⏳ [Novix Core] جاري سحب الأوزان السيادية من المستودع: {MODEL_ID}...")
 try:
-    device = "cuda" if torch.cuda.is_available() else "cpu"
+    # تحميل الأوزان مباشرة إلى البيئة المحلية للـ Space
+    download_model(APP_DIR, repo_id=MODEL_ID, revision="main")
+    print("✅ تم تحميل الأوزان السيادية لـ Novix بنجاح.")
+except Exception as e:
+    print(f"⚠️ تنبيه أثناء تحميل الأوزان: {e}. سيتم الاعتماد على الأوزان المحلية إن وجدت.")
+
+# تهيئة وإقلاع المحرك من الفئة الأصلية المستقرة
+from levo_inference import LeVoInference
+MODEL = None
+
+EXAMPLE_LYRICS = """
+[intro-medium]
+
+[verse]
+随风去流浪
+我不想停留原地
+原地只有无尽循环
+不再规划人生
+不再遵循地图
+
+[chorus]
+让我随风去流浪
+邂逅未知的自己
+生命最绚烂的章节
+""".strip()
+
+# قراءة قاموس التوكنز الصوتي الأصلي للموديل
+vocab_path = op.join(APP_DIR, 'conf/vocab.yaml')
+if op.exists(vocab_path):
+    with open(vocab_path, 'r', encoding='utf-8') as file:
+        STRUCTS = yaml.safe_load(file)
+else:
+    STRUCTS = ['[intro]', '[intro-short]', '[intro-medium]', '[verse]', '[chorus]', '[bridge]', '[inst]', '[inst-short]', '[inst-medium]', '[outro]', '[outro-short]', '[outro-medium]']
+
+def save_as_flac(sample_rate, audio_data):
+    if isinstance(audio_data, tuple):
+        sample_rate, audio_data = audio_data
+        
+    if audio_data.dtype == np.float64:
+        audio_data = audio_data.astype(np.float32)
+        
+    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".flac")
+    sf.write(temp_file, audio_data, sample_rate, format='FLAC')
+    return temp_file.name
+
+# دالة التوليد الأصلية بعد ربطها بـ Novix Core
+def generate_song(lyric, description=None, prompt_audio=None, genre=None, cfg_coef=None, temperature=0.1, top_k=-1, gen_type="mixed", progress=gr.Progress(track_tqdm=True)):
+    global MODEL
+    global STRUCTS
     
-    print("🧠 جاري شحن المفسر النصي الأصلي لـ Qwen2...")
-    # الشحن من المستودع القياسي لضمان استقرار ملفات الـ Tokenizer
-    tokenizer = AutoTokenizer.from_pretrained("Qwen/Qwen2-7B", trust_remote_code=True)
+    if MODEL is None:
+        return None, json.dumps({"error": "المحرك لم يتم تحميله في الذاكرة بعد. يرجى إعادة تحديث الصفحة أو مراجعة السجلات."})
+        
+    params = {'cfg_coef': cfg_coef, 'temperature': temperature, 'top_k': top_k}
+    params = {k: v for k, v in params.items() if v is not None}
+    vocal_structs = ['[verse]', '[chorus]', '[bridge]']
+    sample_rate = MODEL.cfg.sample_rate
     
-    print("⚡ جاري صهر مصفوفات Novix الـ 32GB في الذاكرة الحية...")
-    model = AutoModelForCausalLM.from_pretrained(
-        MODEL_ID, 
-        torch_dtype=torch.float32,
-        low_cpu_mem_usage=True, 
-        trust_remote_code=True
-    ).to(device)
+    # تنسيق وتطهير الكلمات والمقاطع الهيكلية
+    lyric = lyric.replace("[intro]", "[intro-short]").replace("[inst]", "[inst-short]").replace("[outro]", "[outro-short]")
+    paragraphs = [p.strip() for p in lyric.strip().split('\n\n') if p.strip()]
     
-    print("✅ النصر! المحرك شغال ومربوط بالمصفوفات الحقيقية 100%.")
-except Exception as e:
-    print(f"❌ فشل شحن المحرك: {e}")
-    model = None
-    tokenizer = None
-
-def generate_music_independent(lyric, genre, description, cfg_coef, temperature):
-    try:
-        if model is None or tokenizer is None:
-            return None, {"error": "المحرك في وضع الخمول. تأكدي من سلامة الملفات وسجلات الحاوية."}
-            
-        output_path = os.path.join(APP_DIR, "output_song.wav")
-        print("🧠 جاري صهر المدخلات وتوليد التوكنز اللحنية أوفلاين...")
+    if len(paragraphs) < 1:
+        return None, json.dumps("Lyrics can not be left blank")
         
-        # صياغة التلقين بالشكل الهيكلي لـ LeVo
-        full_prompt = f"<lyric>{lyric}</lyric>"
-        if description and description.strip() != "":
-            full_prompt = f"<description>{description}</description> " + full_prompt
-            
-        inputs = tokenizer(full_prompt, return_tensors="pt").to(model.device)
+    paragraphs_norm = []
+    vocal_flag = False
+    
+    for para in paragraphs:
+        lines = para.splitlines()
+        struct_tag = lines[0].strip().lower()
         
-        # إطلاق عملية التوليد الفعلي من أوزان الـ 32GB
-        with torch.no_grad():
-            output_tokens = model.generate(
-                **inputs,
-                max_new_tokens=256,
-                temperature=float(temperature),
-                do_sample=True
-            )
+        if struct_tag not in STRUCTS:
+            return None, json.dumps(f"Segments should start with a structure tag in {STRUCTS}")
             
-        print("🎵 معالجة المصفوفات وحفظ دفق الصوت الحقيقي...")
-        generated_data = output_tokens[0].cpu().float().numpy()
+        if struct_tag in vocal_structs:
+            vocal_flag = True
+            if len(lines) < 2 or not [line.strip() for line in lines[1:] if line.strip()]:
+                return None, json.dumps("The following segments require lyrics: [verse], [chorus], [bridge]")
+            else:
+                new_para_list = []
+                for line in lines[1:]:
+                    new_para_list.append(re.sub(r"[^\w\s\[\]\-\u4e00-\u9fff\u3040-\u309f\u30a0-\u30ff\uac00-\ud7af\u00c0-\u017f]", "", line))
+                new_para_str = f"{struct_tag} {'.'.join(new_para_list)}"
+        else:
+            if len(lines) > 1:
+                return None, json.dumps("The following segments should not contain lyrics: [intro], [intro-short], [intro-medium], [inst], [inst-short], [inst-medium], [outro], [outro-short], [outro-medium]")
+            else:
+                new_para_str = struct_tag
+        paragraphs_norm.append(new_para_str)
         
-        # تسوية الموجة الصوتية خطياً PCM لمنع الانهيار
-        audio_waveform = generated_data / (np.max(np.abs(generated_data)) + 1e-5)
+    if not vocal_flag:
+        return None, json.dumps(f"The lyric must contain at least one of the following structures: {vocal_structs}")
         
-        # حفظ الأغنية بالتردد القياسي للموديل
-        torchaudio.save(output_path, torch.tensor(audio_waveform).unsqueeze(0), 32000)
-        
-        return output_path, {"status": "🎯 تم التوليد بنجاح سيادي كامل!", "device": str(model.device)}
-        
-    except Exception as err:
-        print(f"❌ خطأ أثناء التوليد: {str(err)}")
-        return None, {"error": str(err)}
+    lyric_norm = " ; ".join(paragraphs_norm)
+
+    if prompt_audio is not None:
+        genre = None
+        description = None
+    elif description is not None and description != "":
+        genre = None
+        if description[-1] != ".":
+            description = description + "."
+
+    progress(0.0, "⚡ [Novix Core] التوليد مستمر الآن...")
+    start = time.time()
+    
+    # تشغيل المصفوفات الحقيقية للأوزان المستقلة
+    prompt_path = op.join(APP_DIR, "tools/new_prompt.pt")
+    audio_data = MODEL(lyric_norm, description, prompt_audio, genre, prompt_path, gen_type, params).cpu().permute(1, 0).float().numpy()
+
+    end = time.time()
+    
+    input_config = {
+        "lyric": lyric_norm,
+        "genre": genre,
+        "prompt_audio": prompt_audio,
+        "description": description,
+        "params": params,
+        "inference_duration": end - start,
+        "timestamp": datetime.now().isoformat(),
+        "engine": "Novix Sovereign Studio (Independent Mode)"
+    }
+    
+    filepath = save_as_flac(sample_rate, audio_data)
+    return filepath, json.dumps(input_config, indent=2)
 
-# بناء الواجهة الرسومية الثابتة والمستقرة لـ Gradio
-with gr.Blocks(title="Novix Sovereign Studio") as demo:
+
+# بناء الواجهة الاحترافية الكبرى لـ Gradio
+with gr.Blocks(title="Novix Sovereign Studio Pro") as demo:
     gr.Markdown("# 🎵 استوديو Novix المستقل والمملوك لك بالكامل 100%")
-    gr.Markdown("🛡️ تم إصلاح الكود بالكامل ودمج مفسر Qwen2 المستقر لوضح النهار.")
+    gr.Markdown("🛡️ تم فك الارتباط من خوادم الشر��ات وتوجيه النواة لأوزانكِ الخاصة للإنتاج والربح الحر.")
     
     with gr.Row():
         with gr.Column():
-            lyric = gr.Textbox(label="Lyrics", lines=5, value="[intro]\n[verse]\n随风去流浪...")
-            genre = gr.Radio(choices=["Auto", "Pop", "Rock", "Ballad"], label="Genre Select", value="Auto")
-            description = gr.Textbox(label="Song Description (Optional)", placeholder="female, sad pop, piano")
-            cfg_coef = gr.Slider(label="CFG Coefficient", minimum=0.1, maximum=3.0, value=1.8)
-            temperature = gr.Slider(label="Temperature", minimum=0.1, maximum=2.0, value=0.8)
+            lyric = gr.Textbox(
+                label="Lyrics",
+                lines=5,
+                max_lines=15,
+                value=EXAMPLE_LYRICS,
+                info="قوالب المقاطع الصوتية المدعومة: [intro], [verse], [chorus], [bridge], [inst], [outro]"
+            )
+
+            with gr.Tabs(elem_id="extra-tabs"):
+                with gr.Tab("Genre Select"):
+                    genre = gr.Radio(
+                        choices=["Auto", "Pop", "Latin", "Rock", "Electronic", "Metal", "Country", "R&B/Soul", "Ballad", "Jazz", "World", "Hip-Hop", "Funk", "Soundtrack"],
+                        label="Genre Select (Optional)",
+                        value="Auto",
+                        interactive=True
+                    )
+                with gr.Tab("Text Prompt"):
+                    description = gr.Textbox(
+                        label="Song Description (Optional)",
+                        info="اكتبي مواصفات الصوت بالأرقام أو الإنجليزية (مثال: female, sad pop, piano).",
+                        placeholder="female, rock, motivational, electric guitar, bass guitar, drum kit.",
+                        lines=1,
+                        max_lines=2
+                    )
+                with gr.Tab("Audio Prompt"):
+                    prompt_audio = gr.Audio(
+                        label="Prompt Audio (Optional)",
+                        type="filepath"
+                    )
+
+            with gr.Accordion("Advanced Config", open=False):
+                cfg_coef = gr.Slider(label="CFG Coefficient", minimum=0.1, maximum=3.0, step=0.1, value=1.8, interactive=True)
+                temperature = gr.Slider(label="Temperature", minimum=0.1, maximum=2.0, step=0.1, value=0.8, interactive=True)
+                
+            with gr.Row():
+                generate_btn = gr.Button("Generate Song (Sovereign Mode)", variant="primary")
+        
         with gr.Column():
             output_audio = gr.Audio(label="Generated Song", type="filepath")
-            output_json = gr.JSON(label="System Info")
-
-    generate_btn = gr.Button("Generate Song (Sovereign Mode)", variant="primary")
+            output_json = gr.JSON(label="Generated Info")
+    
     generate_btn.click(
-        fn=generate_music_independent, 
-        inputs=[lyric, genre, description, cfg_coef, temperature], 
+        fn=generate_song,
+        inputs=[lyric, description, prompt_audio, genre, cfg_coef, temperature, gr.State(5000)],
         outputs=[output_audio, output_json]
     )
 
+# تشغيل الإقلاع المستقل للنواة
 if __name__ == "__main__":
-    demo.launch()
+    torch.set_num_threads(1)
+    ckpt_path = op.join(APP_DIR, "ckpt")
+    
+    # التأكد من وجود مجلد الأوزان وإقلاع النموذج فوراً
+    if not op.exists(ckpt_path):
+        os.makedirs(ckpt_path, exist_ok=True)
+        
+    print("🧠 جاري صهر وبناء بيئة الاستدلال الصوتي المستقل...")
+    MODEL = LeVoInference(ckpt_path)
+    print("✅ النصر الكلي! الاستوديو شغال أوفلاين وجاهز لاستقبال ضربة زر التوليد.")
+    
+    demo.launch(server_name="0.0.0.0", server_port=7860)