app.py

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 soundfile as sf
import numpy as np
import tempfile

from download import download_model

# 🎯 خطوة السيادة الأولى: توجيه المحرك لتحميل أوزانكِ الخاصة بدلاً من المستودع العام
APP_DIR = op.dirname(op.abspath(__file__))
MODEL_ID = "Novix/SongGenerationtwo"  # مستودع أوزانكِ السيادية

print(f"⏳ [Novix Core] جاري سحب الأوزان السيادية من المستودع: {MODEL_ID}...")
try:
    # تحميل الأوزان مباشرة إلى البيئة المحلية للـ 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
    
    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
    
    # تنسيق وتطهير الكلمات والمقاطع الهيكلية
    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()]
    
    if len(paragraphs) < 1:
        return None, json.dumps("Lyrics can not be left blank")
        
    paragraphs_norm = []
    vocal_flag = False
    
    for para in paragraphs:
        lines = para.splitlines()
        struct_tag = lines[0].strip().lower()
        
        if struct_tag not in STRUCTS:
            return None, json.dumps(f"Segments should start with a structure tag in {STRUCTS}")
            
        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)
        
    if not vocal_flag:
        return None, json.dumps(f"The lyric must contain at least one of the following structures: {vocal_structs}")
        
    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 Pro") as demo:
    gr.Markdown("# 🎵 استوديو Novix المستقل والمملوك لك بالكامل 100%")
    gr.Markdown("🛡️ تم فك الارتباط من خوادم الشركات وتوجيه النواة لأوزانكِ الخاصة للإنتاج والربح الحر.")
    
    with gr.Row():
        with gr.Column():
            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="Generated Info")
    
    generate_btn.click(
        fn=generate_song,
        inputs=[lyric, description, prompt_audio, genre, cfg_coef, temperature, gr.State(5000)],
        outputs=[output_audio, output_json]
    )

# تشغيل الإقلاع المستقل للنواة
if __name__ == "__main__":
    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)