import warnings warnings.filterwarnings("ignore", category=FutureWarning) import os, json, logging, urllib.parse, random import google.generativeai as genai from faster_whisper import WhisperModel from dotenv import load_dotenv import prompts import intelligence_utils logger = logging.getLogger(__name__) load_dotenv() # Server-level fallback key (optional — can be empty if users bring their own) _SERVER_API_KEY = os.getenv("GEMINI_API_KEY", "") class VocalisIntelligence: def __init__(self): logger.info("Initializing Whisper (medium)...") self.whisper = WhisperModel("medium", device="cpu", compute_type="int8") self.model_name = intelligence_utils.probe_available_models() logger.info(f"AI Brain Active: {self.model_name}") async def analyze_audio(self, audio_path: str, api_key: str = None): try: segments, info = self.whisper.transcribe(audio_path, beam_size=5, word_timestamps=True) # Collect all words with precise timestamps all_words = [] for seg in segments: if seg.words: for w in seg.words: all_words.append({"start": round(w.start, 2), "end": round(w.end, 2), "word": w.word.strip()}) elif seg.text.strip(): all_words.append({"start": round(seg.start, 2), "end": round(seg.end, 2), "word": seg.text.strip()}) # Group words into natural phrases (5-8 words each) transcript_data = [] phrase_words = [] phrase_start = 0.0 for i, w in enumerate(all_words): # Force a new phrase if there's a big time gap (silence between words) if phrase_words and (w["start"] - all_words[i-1]["end"]) > 1.5: transcript_data.append({"start": phrase_start, "text": " ".join(phrase_words)}) phrase_words = [] if not phrase_words: phrase_start = w["start"] phrase_words.append(w["word"]) # Split at natural breaks: punctuation or ~12 words ends_sentence = w["word"][-1] in '.!?,' if w["word"] else False if ends_sentence or len(phrase_words) >= 12: transcript_data.append({"start": phrase_start, "text": " ".join(phrase_words)}) phrase_words = [] # Don't forget remaining words if phrase_words: transcript_data.append({"start": phrase_start, "text": " ".join(phrase_words)}) logger.info(f"Whisper detected: lang={info.language}, prob={info.language_probability:.2f}, words={len(all_words)}, phrases={len(transcript_data)}") if transcript_data: logger.info(f"First 5 phrases: {transcript_data[:5]}") else: logger.warning("Whisper returned ZERO words — audio may be instrumental") raw_text = "\n".join([f"[{s['start']:.2f}s]: {s['text']}" for s in transcript_data]) # Use per-request key if provided, fall back to server key effective_key = api_key or _SERVER_API_KEY if not effective_key: raise ValueError("No Gemini API key provided. Please enter your API key in the settings panel.") genai.configure(api_key=effective_key) model = genai.GenerativeModel(self.model_name) prompt = f"{prompts.SYSTEM_PROMPT}\n\nLYRIC TRANSCRIPT:\n{raw_text}" response = model.generate_content(prompt, generation_config={"response_mime_type": "application/json"}) try: result = json.loads(intelligence_utils.cleanup_json_response(response.text)) except: result = intelligence_utils.get_fallback_result(transcript_data) clean_prompt = result.get("metadata", {}).get("coverArtPrompt", "Abstract cinematic") seed = random.randint(0, 999999) result["metadata"]["coverArtUrl"] = f"https://image.pollinations.ai/prompt/{urllib.parse.quote(clean_prompt)}?width=1024&height=1024&seed={seed}" return result except Exception as e: logger.error(f"Analysis Failed: {e}"); raise e director = VocalisIntelligence()