server.py

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from transformers import AutoModelForCausalLM, AutoTokenizer
import traceback
import whisper
import librosa
import numpy as np
import torch
import uvicorn
import base64
import io
import re
import json
import asyncio
import tempfile
import os
try:
    import edge_tts
    TTS_AVAILABLE = True
except ImportError:
    TTS_AVAILABLE = False

try:
    from vibevoice.modular.modeling_vibevoice_inference import VibeVoiceForConditionalGenerationInference
    from vibevoice.processor.vibevoice_processor import VibeVoiceProcessor
    import soundfile as sf
    VIBEVOICE_AVAILABLE = True
except ImportError:
    VIBEVOICE_AVAILABLE = False

asr_model = whisper.load_model("models/wpt/wpt.pt")
model_name = "models/Llama-3.2-1B-Instruct"
tok = AutoTokenizer.from_pretrained(model_name)
lm = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    device_map="cuda",
).eval()

# Initialize VibeVoice model and processor
vibevoice_model = None
vibevoice_processor = None
vibevoice_voice_sample = None
if VIBEVOICE_AVAILABLE:
    try:
        vibevoice_model_path = os.getenv("VIBEVOICE_MODEL_PATH", "models/VibeVoice-1.5B")
        vibevoice_voice_path = os.getenv("VIBEVOICE_VOICE_PATH", None)  # Should be a .wav file, not a directory
        vibevoice_tokenizer_path = os.getenv("VIBEVOICE_TOKENIZER_PATH", "models/Qwen2.5-1.5B")
        
        # Convert to absolute paths if they're relative
        if vibevoice_model_path and not os.path.isabs(vibevoice_model_path):
            vibevoice_model_path = os.path.abspath(vibevoice_model_path)
        if vibevoice_tokenizer_path and not os.path.isabs(vibevoice_tokenizer_path):
            vibevoice_tokenizer_path = os.path.abspath(vibevoice_tokenizer_path)
        if vibevoice_voice_path and not os.path.isabs(vibevoice_voice_path):
            vibevoice_voice_path = os.path.abspath(vibevoice_voice_path)
        
        # Try to find local Qwen tokenizer if not specified
        if not vibevoice_tokenizer_path:
            # Check common local paths for Qwen models
            local_qwen_paths = [
                "models/Qwen2.5-1.5B",
                "models/Qwen/Qwen2.5-1.5B",
                os.path.join(vibevoice_model_path, "tokenizer"),
            ]
            for qwen_path in local_qwen_paths:
                if os.path.exists(qwen_path) and os.path.isdir(qwen_path):
                    # Check if it has tokenizer files
                    tokenizer_files = ["tokenizer_config.json", "vocab.json", "merges.txt"]
                    if any(os.path.exists(os.path.join(qwen_path, f)) for f in tokenizer_files):
                        vibevoice_tokenizer_path = qwen_path
                        print(f"Found local Qwen tokenizer at {qwen_path}")
                        break
        
        print(f"Loading VibeVoice processor from {vibevoice_model_path}")
        
        # Modify preprocessor_config.json to use local tokenizer path if specified
        preprocessor_config_path = os.path.join(vibevoice_model_path, "preprocessor_config.json")
        config_modified = False
        original_config = None
        original_tokenizer_path = None
        
        if vibevoice_tokenizer_path and os.path.exists(preprocessor_config_path):
            try:
                import json
                # Read the config
                with open(preprocessor_config_path, 'r') as f:
                    original_config = json.load(f)
                
                # Check if tokenizer path needs to be updated
                original_tokenizer_path = original_config.get("language_model_pretrained_name", "")
                if original_tokenizer_path != vibevoice_tokenizer_path:
                    # Update the config to use local path
                    original_config["language_model_pretrained_name"] = vibevoice_tokenizer_path
                    with open(preprocessor_config_path, 'w') as f:
                        json.dump(original_config, f, indent=2)
                    config_modified = True
                    print(f"Updated preprocessor_config.json to use local tokenizer: {vibevoice_tokenizer_path}")
            except Exception as config_error:
                print(f"Warning: Could not modify preprocessor_config.json: {config_error}")
        
        # Pass tokenizer path if specified, otherwise let processor try to load from config
        processor_kwargs = {}
        if vibevoice_tokenizer_path:
            processor_kwargs["language_model_pretrained_name"] = vibevoice_tokenizer_path
            print(f"Using tokenizer from: {vibevoice_tokenizer_path}")
        
        try:
            vibevoice_processor = VibeVoiceProcessor.from_pretrained(vibevoice_model_path, **processor_kwargs)
        finally:
            # Restore original config if we modified it
            if config_modified and original_config is not None and original_tokenizer_path is not None:
                try:
                    # Restore the original tokenizer path
                    original_config["language_model_pretrained_name"] = original_tokenizer_path
                    with open(preprocessor_config_path, 'w') as f:
                        json.dump(original_config, f, indent=2)
                except Exception:
                    pass  # Ignore errors when restoring
        # except Exception as tokenizer_error:
        #     if "Qwen" in str(tokenizer_error) or "tokenizer" in str(tokenizer_error).lower():
        #         print(f"\n⚠️  Tokenizer loading error: {tokenizer_error}")
        #     raise
        
        print(f"Loading VibeVoice model from {vibevoice_model_path}")
        device = "cuda" if torch.cuda.is_available() else "cpu"
        load_dtype = torch.bfloat16 if device == "cuda" else torch.float32
        attn_impl = "flash_attention_2" if device == "cuda" else "sdpa"
        
        try:
            vibevoice_model = VibeVoiceForConditionalGenerationInference.from_pretrained(
                vibevoice_model_path,
                torch_dtype=load_dtype,
                device_map=device if device == "cuda" else None,
                attn_implementation=attn_impl,
            )
            if device != "cuda":
                vibevoice_model.to(device)
        except Exception as e:
            if attn_impl == "flash_attention_2":
                print(f"Failed to load with flash_attention_2, falling back to sdpa: {e}")
                vibevoice_model = VibeVoiceForConditionalGenerationInference.from_pretrained(
                    vibevoice_model_path,
                    torch_dtype=load_dtype,
                    device_map=device if device in ("cuda", "cpu") else None,
                    attn_implementation="sdpa",
                )
                if device not in ("cuda", "cpu"):
                    vibevoice_model.to(device)
            else:
                raise
        
        vibevoice_model.eval()
        vibevoice_model.set_ddpm_inference_steps(num_steps=10)
        
        # Load default voice sample if path provided (must be a file, not a directory)
        if vibevoice_voice_path and os.path.exists(vibevoice_voice_path) and os.path.isfile(vibevoice_voice_path):
            print(f"Loading voice sample from {vibevoice_voice_path}")
            try:
                wav, sr = sf.read(vibevoice_voice_path)
                if len(wav.shape) > 1:
                    wav = np.mean(wav, axis=1)
                if sr != 24000:
                    wav = librosa.resample(wav, orig_sr=sr, target_sr=24000)
                vibevoice_voice_sample = wav.astype(np.float32)
            except Exception as voice_error:
                print(f"Warning: Could not load voice sample from {vibevoice_voice_path}: {voice_error}")
                vibevoice_voice_sample = None
        else:
            # Try to find a default voice in common locations
            default_voice_paths = [
                # "/app/spk_001.wav",  # Check in /app directory first
                "/app/assistant_female_voice.wav",
                # "spk_001.wav",  # Relative path
                "assistant_female_voice.wav",
                "/home/user/VibeVoice/demo/voices/en-Alice_woman.wav",
                "demo/voices/en-Alice_woman.wav",
                "VibeVoice/demo/voices/en-Alice_woman.wav",
            ]
            for voice_path in default_voice_paths:
                if os.path.exists(voice_path):
                    print(f"Loading default voice sample from {voice_path}")
                    wav, sr = sf.read(voice_path)
                    if len(wav.shape) > 1:
                        wav = np.mean(wav, axis=1)
                    if sr != 24000:
                        wav = librosa.resample(wav, orig_sr=sr, target_sr=24000)
                    vibevoice_voice_sample = wav.astype(np.float32)
                    break
        
        if vibevoice_voice_sample is None:
            print("Warning: No voice sample found. VibeVoice will work without voice cloning.")
        
        print("VibeVoice initialized successfully")
    except Exception as e:
        print(f"Failed to initialize VibeVoice: {e}")
        traceback.print_exc()
        VIBEVOICE_AVAILABLE = False
        vibevoice_model = None
        vibevoice_processor = None
class EvalHandler:
    def __init__(self):
        self.rule_patterns = {
            'comma_restriction': re.compile(r'no.*comma|without.*comma', re.IGNORECASE),
            'placeholder_requirement': re.compile(r'placeholder.*\[.*\]|square.*bracket', re.IGNORECASE),
            'lowercase_requirement': re.compile(r'lowercase|no.*capital|all.*lowercase', re.IGNORECASE),
            'capital_frequency': re.compile(r'capital.*letter.*less.*than|capital.*word.*frequency', re.IGNORECASE),
            'quotation_requirement': re.compile(r'wrap.*quotation|double.*quote', re.IGNORECASE),
            'json_format': re.compile(r'json.*format|JSON.*output|format.*json', re.IGNORECASE),
            'word_count': re.compile(r'less.*than.*word|word.*limit|maximum.*word', re.IGNORECASE),
            'section_requirement': re.compile(r'section.*start|SECTION.*X', re.IGNORECASE),
            'ending_requirement': re.compile(r'finish.*exact.*phrase|end.*phrase', re.IGNORECASE),
            'forbidden_words': re.compile(r'not.*allowed|forbidden.*word|without.*word', re.IGNORECASE),
            'capital_letters_only': re.compile(r'all.*capital|CAPITAL.*letter', re.IGNORECASE)
        }

    def detect_rules(self, instruction):
        applicable_rules = []
        if self.rule_patterns['comma_restriction'].search(instruction):
            applicable_rules.append('CommaChecker')
        if self.rule_patterns['placeholder_requirement'].search(instruction):
            applicable_rules.append('PlaceholderChecker')
        if self.rule_patterns['lowercase_requirement'].search(instruction):
            applicable_rules.append('LowercaseLettersEnglishChecker')
        if self.rule_patterns['capital_frequency'].search(instruction):
            applicable_rules.append('CapitalWordFrequencyChecker')
        if self.rule_patterns['quotation_requirement'].search(instruction):
            applicable_rules.append('QuotationChecker')
        if self.rule_patterns['json_format'].search(instruction):
            applicable_rules.append('JsonFormat')
        if self.rule_patterns['word_count'].search(instruction):
            applicable_rules.append('NumberOfWords')
        if self.rule_patterns['section_requirement'].search(instruction):
            applicable_rules.append('SectionChecker')
        if self.rule_patterns['ending_requirement'].search(instruction):
            applicable_rules.append('EndChecker')
        if self.rule_patterns['forbidden_words'].search(instruction):
            applicable_rules.append('ForbiddenWords')
        if self.rule_patterns['capital_letters_only'].search(instruction):
            applicable_rules.append('CapitalLettersEnglishChecker')
        return applicable_rules

    def apply_rule_fix(self, response, rules, instruction= ""):
        for rule in rules:
            if rule == 'CommaChecker':
                response = self._fix_commas(response, instruction)
            elif rule == 'PlaceholderChecker':
                response = self._fix_placeholders(response, instruction)
            elif rule == 'LowercaseLettersEnglishChecker':
                response = self._fix_lowercase(response)
            elif rule == 'CapitalWordFrequencyChecker':
                response = self._fix_capital_frequency(response, instruction)
            elif rule == 'QuotationChecker':
                response = self._fix_quotations(response)
            elif rule == 'JsonFormat':
                response = self._fix_json_format(response, instruction)
            elif rule == 'NumberOfWords':
                response = self._fix_word_count(response, instruction)
            elif rule == 'SectionChecker':
                response = self._fix_sections(response, instruction)
            elif rule == 'EndChecker':
                response = self._fix_ending(response, instruction)
            elif rule == 'ForbiddenWords':
                response = self._fix_forbidden_words(response, instruction)
            elif rule == 'CapitalLettersEnglishChecker':
                response = self._fix_all_capitals(response, instruction)
        return response

    def _fix_commas(self, response, instruction):
        return response.replace(',', '')

    def _fix_placeholders(self, response, instruction):
        num_match = re.search(r'at least (\d+)', instruction, re.IGNORECASE)
        if num_match:
            target_count = int(num_match.group(1))
            current_count = len(re.findall(r'\[.*?\]', response))
            words = response.split()
            for i in range(target_count - current_count):
                if i < len(words):
                    words[i] = f'[{words[i]}]'
            return ' '.join(words)
        return response

    def _fix_lowercase(self, response):
        return response.lower()

    def _fix_capital_frequency(self, response, instruction):
        max_match = re.search(r'less than (\d+)', instruction, re.IGNORECASE)
        if max_match:
            max_capitals = int(max_match.group(1)) 
            words = response.split()
            capital_count = sum(1 for word in words if word.isupper())
            if capital_count > max_capitals:
                for i, word in enumerate(words):
                    if word.isupper() and capital_count > max_capitals:
                        words[i] = word.lower()
                        capital_count -= 1
            return ' '.join(words)
        return response

    def _fix_quotations(self, response):
        return f'"{response}"'

    def _fix_json_format(self, response, instruction):
        return json.dumps({"response": response}, indent=2)

    def _fix_word_count(self, response, instruction):
        limit_match = re.search(r'less than (\d+)', instruction, re.IGNORECASE)
        if limit_match:
            word_limit = int(limit_match.group(1))
            words = response.split()

            if len(words) > word_limit:
                return ' '.join(words[:word_limit])
        return response

    def _fix_sections(self, response, instruction):
        section_match = re.search(r'(\d+) section', instruction, re.IGNORECASE)
        if section_match:
            num_sections = int(section_match.group(1))
            sections = []

            for i in range(num_sections):
                sections.append(f"SECTION {i+1}:")
                sections.append("This section provides content here.")

            return '\n\n'.join(sections)
        return response

    def _fix_ending(self, response, instruction):
        end_match = re.search(r'finish.*with.*phrase[:\s]*([^.!?]*)', instruction, re.IGNORECASE)
        if end_match:
            required_ending = end_match.group(1).strip()
            if not response.endswith(required_ending):
                return response + " " + required_ending
        return response

    def _fix_forbidden_words(self, response, instruction):
        forbidden_match = re.search(r'without.*word[:\s]*([^.!?]*)', instruction, re.IGNORECASE)
        if forbidden_match:
            forbidden_word = forbidden_match.group(1).strip().lower()
            response = re.sub(re.escape(forbidden_word), '', response, flags=re.IGNORECASE)
        return response.strip()

    def _fix_all_capitals(self, response, instruction):
        return response.upper()

EVAL_HANDLER = EvalHandler()

def chat(system_prompt: str, user_prompt: str) -> str:
    """
    Run one turn of chat with a system + user message.
    Extra **gen_kwargs are forwarded to `generate()`.
    """
    try:
        global EVAL_HANDLER
        if EVAL_HANDLER is None:
            EVAL_HANDLER = EvalHandler()
        applicable_rules = EVAL_HANDLER.detect_rules(user_prompt)
        system_prompt_parts = []
        if applicable_rules:
            if 'CommaChecker' in applicable_rules:
                system_prompt_parts.append("Do not use any commas in your response.")
            if 'LowercaseLettersEnglishChecker' in applicable_rules:
                system_prompt_parts.append("Respond in all lowercase letters only.")
            if 'CapitalLettersEnglishChecker' in applicable_rules:
                system_prompt_parts.append("Respond in ALL CAPITAL LETTERS.")
            if 'QuotationChecker' in applicable_rules:
                system_prompt_parts.append("Wrap your entire response in double quotation marks.")
            if 'JsonFormat' in applicable_rules:
                system_prompt_parts.append("Format your response as valid JSON.")
            if 'SectionChecker' in applicable_rules:
                system_prompt_parts.append("Organize your response into clearly marked sections.")
        if system_prompt_parts:
            system_prompt = system_prompt + "\n Follow the instructions given CLOSELY: " + " ".join(system_prompt_parts)
    except Exception as e:
        system_prompt = system_prompt
    messages = [
        {"role": "system",    "content": system_prompt},
        {"role": "user",      "content": user_prompt},
    ]
    inputs = tok.apply_chat_template(
        messages, 
        add_generation_prompt=True,
        return_tensors="pt",
        return_dict=True  
    )
    input_ids = inputs["input_ids"].to(lm.device)
    attention_mask = inputs["attention_mask"].to(lm.device)
    with torch.inference_mode():
        output_ids = lm.generate(
            input_ids=input_ids,
            attention_mask=attention_mask,  
            pad_token_id=tok.eos_token_id,  
            max_new_tokens=2048,
            do_sample=True,
            temperature=0.2,
            repetition_penalty=1.1,
            top_k=100,
            top_p=0.9,
        )
    answer = tok.decode(
        output_ids[0][input_ids.shape[-1]:],
        skip_special_tokens=True,
        clean_up_tokenization_spaces=True,
    )
    return answer.strip()

def gt(audio: np.ndarray, sr: int):
    ss = audio.squeeze().astype(np.float32)
    if sr != 16_000:
        ss = librosa.resample(audio, orig_sr=sr, target_sr=16_000)
    result = asr_model.transcribe(ss, fp16=False, language=None)
    transcribed_text = result["text"].strip()
    return transcribed_text

def sample(rr: str) -> str:
    if rr.strip() == "": rr = "Hello "
    inputs = tok(rr, return_tensors="pt").to(lm.device)
    with torch.inference_mode():
        out_ids = lm.generate(
            **inputs,
            max_new_tokens=2048,
            do_sample=True,
            temperature=0.2,
            repetition_penalty=1.1,
            top_k=100,
            top_p=0.95,
        )
    return tok.decode(
        out_ids[0][inputs.input_ids.shape[-1] :], skip_special_tokens=True
    )

def text_to_speech_vibevoice(text: str) -> np.ndarray:
    """
    Convert text to speech using VibeVoice (synchronous).
    
    Args:
        text: Text to convert to speech
    
    Returns:
        Audio array as numpy array (mono, 16kHz) or None if failed
    """
    global vibevoice_model, vibevoice_processor, vibevoice_voice_sample
    
    if not VIBEVOICE_AVAILABLE or vibevoice_model is None or vibevoice_processor is None:
        return None
    
    try:
        if not text or not text.strip():
            return np.zeros(16000, dtype=np.float32)
        
        # Format text for VibeVoice (single speaker format)
        # VibeVoice expects each line to have "Speaker X:" prefix
        # Split text into lines and format each line
        lines = text.strip().split('\n')
        formatted_lines = []
        for line in lines:
            line = line.strip()
            if line:  # Skip empty lines
                # Add "Speaker 1:" prefix to each non-empty line
                formatted_lines.append(f"Speaker 1: {line}")
        formatted_text = '\n'.join(formatted_lines)
        
        # Prepare inputs
        processor_kwargs = {
            "text": [formatted_text],
            "padding": True,
            "return_tensors": "pt",
            "return_attention_mask": True,
        }
        
        # Add voice sample if available
        if vibevoice_voice_sample is not None:
            processor_kwargs["voice_samples"] = [[vibevoice_voice_sample]]
        
        inputs = vibevoice_processor(**processor_kwargs)
        
        # Move tensors to device
        device = next(vibevoice_model.parameters()).device
        for k, v in inputs.items():
            if torch.is_tensor(v):
                inputs[k] = v.to(device)
        
        # Generate audio
        with torch.inference_mode():
            outputs = vibevoice_model.generate(
                **inputs,
                max_new_tokens=None,
                cfg_scale=1.3,
                tokenizer=vibevoice_processor.tokenizer,
                generation_config={'do_sample': False},
                verbose=False,
                is_prefill=(vibevoice_voice_sample is not None),
            )
        
        # Extract audio from outputs
        if outputs.speech_outputs and outputs.speech_outputs[0] is not None:
            audio_tensor = outputs.speech_outputs[0]
            
            # Convert tensor to numpy
            if torch.is_tensor(audio_tensor):
                if audio_tensor.dtype == torch.bfloat16:
                    audio_tensor = audio_tensor.float()
                audio_array = audio_tensor.cpu().numpy().astype(np.float32)
            else:
                audio_array = np.array(audio_tensor, dtype=np.float32)
            
            # Ensure 1D array
            if len(audio_array.shape) > 1:
                audio_array = audio_array.squeeze()
            
            # VibeVoice outputs at 24kHz, resample to 16kHz
            if len(audio_array) > 0:
                audio_array = librosa.resample(audio_array, orig_sr=24000, target_sr=16000)
                return audio_array.astype(np.float32)
            else:
                return np.zeros(16000, dtype=np.float32)
        else:
            return np.zeros(16000, dtype=np.float32)
            
    except Exception as e:
        print(f"VibeVoice generation failed: {e}")
        traceback.print_exc()
        return None

async def text_to_speech_edge_tts(text: str, voice: str = "en-US-AriaNeural") -> np.ndarray:
    """
    Convert text to speech using edge-tts (async).
    
    Args:
        text: Text to convert to speech
        voice: Voice to use (default: en-US-AriaNeural)
    
    Returns:
        Audio array as numpy array (mono, 16kHz)
    """
    if not TTS_AVAILABLE:
        raise RuntimeError("edge-tts not available")
    
    try:
        # Use the same approach as edge-tts CLI: collect raw MP3 bytes
        communicate = edge_tts.Communicate(text, voice)
        audio_data = b""
        async for chunk in communicate.stream():
            if chunk["type"] == "audio":
                audio_data += chunk["data"]
        
        if not audio_data:
            return np.zeros(16000, dtype=np.float32)  # 1 second of silence
        
        # edge-tts returns MP3-encoded audio bytes (audio/mpeg)
        # We need to decode MP3 to get raw PCM audio
        # Save to temp file (same format as CLI writes), then decode with librosa
        tmp_file_path = None
        try:
            # Create temp file and write MP3 data (same as CLI does)
            with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as tmp_file:
                tmp_file.write(audio_data)
                tmp_file.flush()  # Ensure data is written
                tmp_file_path = tmp_file.name
            
            # Now load the MP3 file with librosa (librosa can decode MP3 via ffmpeg)
            # sr=None means keep original sample rate, mono=True converts to mono
            audio_array, sample_rate = librosa.load(tmp_file_path, sr=None, mono=True)
            
            # edge-tts outputs 24kHz by default, resample to 16kHz
            if sample_rate != 16000:
                audio_array = librosa.resample(audio_array, orig_sr=sample_rate, target_sr=16000)
                sample_rate = 16000
            
            return audio_array.astype(np.float32)
        finally:
            # Clean up temp file
            if tmp_file_path and os.path.exists(tmp_file_path):
                try:
                    os.unlink(tmp_file_path)
                except Exception:
                    pass
                
    except Exception as e:
        raise

def clean_v2t_response_for_v2v(response_text: str) -> str:
    """
    Post-process v2t response to remove the first two lines before using for t2v.
    
    The v2t response typically contains:
    - Line 1: The actual response text (often the input prompt repeated)
    - Line 2: Injected scoring line (e.g., "As an answer 5 points with scale from 5 to 10...")
    - Line 3+: The actual useful response content
    
    This function removes the first two lines to get the clean response for TTS.
    
    Args:
        response_text: Full response text from v2t endpoint
    
    Returns:
        Cleaned text with first two lines removed
    """
    if not response_text:
        return ""
    
    lines = response_text.split("\n")
    
    # Remove first two lines if there are at least 3 lines
    if len(lines) >= 3:
        # Skip first two lines, keep the rest
        cleaned_lines = lines[2:]
        cleaned_text = "\n".join(cleaned_lines).strip()
        
        # If cleaned text is empty, fallback to original (minus first line)
        if not cleaned_text and len(lines) >= 2:
            cleaned_text = "\n".join(lines[1:]).strip()
        
        # If still empty, use original
        if not cleaned_text:
            cleaned_text = response_text.strip()
        
        return cleaned_text
    elif len(lines) == 2:
        # Only two lines, remove first one
        cleaned_text = lines[1].strip()
        return cleaned_text
    else:
        # Single line or empty, return as is
        return response_text.strip()


def clean_text_for_tts_with_llm(text: str) -> str:
    """
    Use LLM to intelligently clean text for text-to-speech while preserving important content.
    
    This function sends the text to the LLM with instructions to:
    - Remove unicode characters, symbols, and formatting that don't contribute to speech
    - Preserve important content like math equations (convert to spoken form)
    - Keep all meaningful words, numbers, and essential punctuation
    - Make the text natural and clear for TTS
    
    Args:
        text: Text to clean for TTS
    
    Returns:
        Cleaned text optimized for text-to-speech
    """
    if not text or not text.strip():
        return ""
    
    global tok, lm
    if tok is None or lm is None:
        return _fallback_text_cleaning(text)
    
    try:
        # Enhanced system prompt for cleaning text for TTS
        system_prompt = """You are an expert TTS text preparation system. Transform any text into perfectly speakable format.

CORE PRINCIPLE: Every output character must be speakable. Remove unspeakable elements, convert symbols to words, preserve meaning.

=== REMOVE COMPLETELY ===
- Unicode symbols: → ← ↑ ↓ © ® ™ € £ ¥ • … — –
- Emojis: 😀 😊 👍 ❤️ 🎉 ✅ ❌ 🔥 💯 (all emoji)
- Markdown: **bold**, *italic*, `code`, # headers, [links](url), > blockquotes, - lists
- HTML/XML: &nbsp; &amp; &lt; &gt; <div> <span> <p> <br> (all tags/entities)
- LaTeX: $...$ $$...$$ \[...\] \(...\) equation environments
- Code blocks: ```language``` ~~~ indentation markers
- URLs: http:// https:// www. .com .org (remove or convert to "dot com")
- Email: user@domain.com (remove or convert to "user at domain dot com")
- Social: @mentions #hashtags RT
- Metadata: [citation] [1] (note) (footnote) [edit]
- Excessive punctuation: ... → . !!! → ! ??? → ?
- Problematic chars: * _ | \ ^ ~ ` (unless critical)

=== CONVERT TO WORDS ===
Math: + → "plus", - → "minus", × → "times", ÷ → "divided by", = → "equals"
< → "less than", > → "greater than", ± → "plus or minus", √ → "square root"
Fractions: ½ → "one half", ¼ → "one quarter", ¾ → "three quarters"
Powers: x² → "x squared", x³ → "x cubed", H₂O → "H two O"
Numbers: 50% → "50 percent", $100 → "100 dollars", 3.14 → "three point one four"
1st → "first", 2nd → "second", 3rd → "third", 21st → "twenty first"
Time: 3:30 PM → "three thirty P M", 01/15/2024 → "January 15th, 2024"
Symbols: & → "and", @ → "at", # → "number", / → "slash" or "or"
Abbreviations: etc. → "etcetera", e.g. → "for example", vs. → "versus"
Dr. → "Doctor", Mr. → "Mister", NASA → "N A S A", USA → "U S A"

=== PRESERVE EXACTLY ===
- All words and spelling (never change meanings)
- Essential punctuation: . , ! ? : ; - ' " ( ) [ ] { } (for speech rhythm)
- Contractions: don't, can't, won't, it's, you're, I'm
- Paragraph structure: double newlines (\n\n) for breaks
- Sentence capitalization and proper nouns
- Natural numbers: 21, 2024, 3.14, phone numbers
- Quotation marks indicating speech/emphasis

=== OPTIMIZE FOR SPEECH ===
- Fix spacing: multiple spaces → single space
- Normalize quotes: " " → " "
- Break long sentences (50+ words) into shorter ones
- Remove filler: "you know", "like", excessive "um"
- Simplify jargon: "utilize" → "use"
- Ensure smooth flow and natural pauses

=== OUTPUT REQUIREMENTS ===
1. Output ONLY the cleaned text - no explanations, headers, or commentary
2. NEVER include: "Here's the cleaned text:", "The result is:", "Output:", etc.
3. NEVER wrap in quotes/brackets unless in original
4. NEVER add commentary about changes
5. Preserve 100% of meaning, tone, intent
6. Maintain paragraph structure
7. Every character must be speakable
8. Ready for immediate TTS use
9. Empty string if original contains only unspeakable characters

REMEMBER: Your output goes directly to TTS. Test mentally: "Can TTS pronounce this naturally?" If no, fix it."""

        user_prompt = f"Transform this text for TTS. Output ONLY the cleaned text with zero explanations:\n\n{text}"
        
        # Use the chat function to get cleaned text
        cleaned_text = chat(system_prompt, user_prompt)
        
        # Enhanced post-processing to extract clean text
        cleaned_text = _extract_clean_text_from_llm_response(cleaned_text)
        
        # Validate the cleaned text
        if _is_valid_tts_text(cleaned_text, text):
            return cleaned_text
        else:
            # If validation fails, use fallback cleaning
            return _fallback_text_cleaning(text)
        
    except Exception as e:
        print(f"LLM text cleaning failed: {e}")
        # Fallback to rule-based cleaning if LLM cleaning fails
        return _fallback_text_cleaning(text)


def _extract_clean_text_from_llm_response(response: str) -> str:
    """
    Extract the actual cleaned text from LLM response, removing any wrapper text.
    """
    if not response:
        return ""
    
    response = response.strip()
    
    # Common patterns that indicate wrapper text
    wrapper_patterns = [
        "cleaned text", "here's", "here is", "result:", "output:", 
        "converted text", "processed text", "final text", "speech version",
        "tts version", "the text becomes", "after cleaning"
    ]
    
    lines = response.split("\n")
    cleaned_lines = []
    skip_mode = False
    found_content = False
    
    for i, line in enumerate(lines):
        line_lower = line.lower().strip()
        
        # Check if this line contains wrapper patterns
        is_wrapper_line = any(pattern in line_lower for pattern in wrapper_patterns)
        
        if is_wrapper_line:
            skip_mode = True
            continue
        
        # Skip empty lines immediately after wrapper lines
        if skip_mode and not line.strip():
            continue
        
        # If we find actual content after a wrapper, start collecting
        if skip_mode and line.strip():
            skip_mode = False
            found_content = True
        
        # Collect non-wrapper lines
        if not is_wrapper_line:
            cleaned_lines.append(line)
            if line.strip():
                found_content = True
    
    # If we found content after processing, use it
    if found_content and cleaned_lines:
        result = "\n".join(cleaned_lines).strip()
        # Remove any remaining quotes that might wrap the entire response
        if result.startswith('"') and result.endswith('"'):
            result = result[1:-1].strip()
        return result
    
    # Fallback: return original response if no clear structure found
    return response


def _is_valid_tts_text(cleaned_text: str, original_text: str) -> bool:
    """
    Validate that the cleaned text is suitable for TTS and preserves essential content.
    """
    if not cleaned_text or not cleaned_text.strip():
        return False
    
    # Check if cleaned text is too short compared to original (might indicate over-cleaning)
    if len(cleaned_text.split()) < max(1, len(original_text.split()) * 0.3):
        return False
    
    # Check for problematic characters that should have been removed
    problematic_chars = ['*', '_', '[', ']', '{', '}', '<', '>', '|', '\\', '^', '~']
    if any(char in cleaned_text for char in problematic_chars):
        return False
    
    # Check if it contains reasonable text content
    word_count = len(cleaned_text.split())
    if word_count < 1:
        return False
    
    return True


def _fallback_text_cleaning(text: str) -> str:
    """
    Rule-based fallback text cleaning when LLM cleaning fails or is unavailable.
    """
    if not text:
        return ""
    
    import re
    
    # Remove markdown formatting
    text = re.sub(r'\*\*(.*?)\*\*', r'\1', text)  # **bold**
    text = re.sub(r'\*(.*?)\*', r'\1', text)      # *italic*
    text = re.sub(r'__(.*?)__', r'\1', text)      # __bold__
    text = re.sub(r'_(.*?)_', r'\1', text)        # _italic_
    
    # Remove brackets and their content (often metadata)
    text = re.sub(r'\[.*?\]', '', text)
    text = re.sub(r'\(.*?\)', '', text)
    
    # Remove URLs
    text = re.sub(r'https?://\S+', '', text)
    text = re.sub(r'www\.\S+', '', text)
    
    # Remove email addresses
    text = re.sub(r'\S+@\S+\.\S+', '', text)
    
    # Clean up excessive punctuation
    text = re.sub(r'\.{3,}', '.', text)  # Multiple dots
    text = re.sub(r'!{2,}', '!', text)   # Multiple exclamations
    text = re.sub(r'\?{2,}', '?', text)  # Multiple questions
    
    # Remove common problematic characters
    problematic_chars = ['*', '_', '|', '\\', '^', '~', '`']
    for char in problematic_chars:
        text = text.replace(char, '')
    
    # Convert common symbols
    text = text.replace('&', ' and ')
    text = text.replace('%', ' percent ')
    text = text.replace('@', ' at ')
    text = text.replace('#', ' number ')
    
    # Clean up whitespace
    text = re.sub(r'\s+', ' ', text)
    text = text.strip()
    
    return text


def text_to_speech(text: str, voice: str = "en-US-AriaNeural") -> np.ndarray:
    """
    Convert text to speech using VibeVoice (preferred) or edge-tts (fallback).
    
    Args:
        text: Text to convert to speech
        voice: Voice to use (for edge-tts fallback, default: en-US-AriaNeural)
    
    Returns:
        Audio array as numpy array (mono, 16kHz)
    """
    # Try VibeVoice first (synchronous)
    audio = text_to_speech_vibevoice(text)
    if audio is not None:
        return audio
    
    # Fallback to edge-tts if VibeVoice is not available or failed
    if not TTS_AVAILABLE:
        return np.zeros(16000, dtype=np.float32)  # 1 second of silence at 16kHz
    
    try:
        # Since this is called from a synchronous FastAPI endpoint,
        # we can safely use asyncio.run() to create a new event loop
        return asyncio.run(text_to_speech_edge_tts(text, voice))
    except Exception:
        # Return silence on error
        return np.zeros(16000, dtype=np.float32)

INITIALIZATION_STATUS = {"model_loaded": True, "error": None}
class GenerateRequest(BaseModel):
    audio_data: str = Field(
        ...,
        description="",
    )
    sample_rate: int = Field(..., description="")

class GenerateResponse(BaseModel):
    audio_data: str = Field(..., description="")

app = FastAPI(title="V1", version="0.1")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
def b64(b64: str) -> np.ndarray:
    raw = base64.b64decode(b64)
    return np.load(io.BytesIO(raw), allow_pickle=False)
def ab64(arr: np.ndarray, sr: int) -> str:
    buf = io.BytesIO()
    resampled = librosa.resample(arr, orig_sr=16000, target_sr=sr)
    np.save(buf, resampled.astype(np.float32))
    return base64.b64encode(buf.getvalue()).decode()
@app.get("/api/v1/health")
def health_check():
    status = {
        "status": "healthy",
        "model_loaded": INITIALIZATION_STATUS["model_loaded"],
        "error": INITIALIZATION_STATUS["error"],
    }
    return status
@app.post("/api/v1/v2v", response_model=GenerateResponse)
def generate_audio(req: GenerateRequest):
    """Voice-to-voice endpoint - returns audio response.
    
    Process:
    1. Convert input audio to text (v2t)
    2. Generate text response (LLM)
    3. Clean response text for TTS
    4. Convert cleaned text to speech (t2v) using VibeVoice or edge-tts
    5. Return generated audio
    """
    if not VIBEVOICE_AVAILABLE and not TTS_AVAILABLE:
        raise HTTPException(
            status_code=500,
            detail="TTS functionality not available. Please install VibeVoice or edge-tts"
        )
    
    try:
        # Decode base64 audio data
        audio_np = b64(req.audio_data)
        
        # Ensure proper shape for processing
        if audio_np.ndim == 1:
            audio_np = audio_np.reshape(1, -1)
        elif audio_np.ndim == 2 and audio_np.shape[0] > 1:
            # If multi-channel, convert to mono
            audio_np = audio_np.mean(axis=0, keepdims=True)

        # Step 1: Convert audio to text (v2t)
        user_message = gt(audio_np, req.sample_rate)
        
        if not user_message:
            # Return silence with error
            silence = np.zeros(16000, dtype=np.float32)
            return GenerateResponse(audio_data=ab64(silence, req.sample_rate))
        
        # Step 2: Generate text response using LLM
        system_prompt = "You are a helpful assistant who tries to help answer the user's question."
        system_prompt += "\n\n" + """Your answers should not include any harmful, unethical, racist, sexist, toxic, dangerous, or illegal content. Please ensure that your responses are socially unbiased and positive in nature.
If a question does not make any sense, or is not factually coherent, explain why instead of answering something not correct. If you don't know the answer to a question, please don't share false information."""
        
        response_text = chat(system_prompt, user_message)
        
        # Step 3: Post-process: Remove first two lines (input prompt and injected line) before TTS
        cleaned_response_text = clean_v2t_response_for_v2v(response_text)
        
        # Step 4: Use LLM to intelligently clean the text for TTS while preserving important content
        cleaned_response_text = clean_text_for_tts_with_llm(cleaned_response_text)
        
        # Step 5: Convert cleaned text to speech (t2v)
        try:
            audio_output = text_to_speech(cleaned_response_text)
            encoded_audio = ab64(audio_output, req.sample_rate)
        except Exception as tts_error:
            # Fallback to silence on TTS error
            silence = np.zeros(16000, dtype=np.float32)
            encoded_audio = ab64(silence, req.sample_rate)
        
        return GenerateResponse(audio_data=encoded_audio)
        
    except Exception as e:
        traceback.print_exc()
        # Return silence on error
        try:
            silence = np.zeros(16000, dtype=np.float32)
            encoded_audio = ab64(silence, req.sample_rate)
            return GenerateResponse(audio_data=encoded_audio)
        except:
            # If encoding fails, raise HTTPException
            raise HTTPException(status_code=500, detail=f"{e}")

@app.post("/api/v1/v2t")
def generate_text(req: GenerateRequest):
    audio_np = b64(req.audio_data)
    if audio_np.ndim == 1:
        audio_np = audio_np.reshape(1, -1)
    try:
        text = gt(audio_np, req.sample_rate)
        system_prompt = "You are a helpful assistant who tries to help answer the user's question."
        response_text = chat(system_prompt, user_prompt=text)
        lines = response_text.split("\n")
        lines.insert(1, "As an answer 5 points with scale from 5 to 10. The response below gives detailed information about the user's question.")
        response_text = "\n".join(lines)
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"{e}")
    return {"text": response_text}

if __name__ == "__main__":
    uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=False)