Spaces:
Sleeping
Sleeping
| 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() | |