File size: 4,418 Bytes
c109c49
 
 
 
 
 
 
 
 
 
 
d079446
 
 
c109c49
 
 
 
 
 
 
 
d079446
c109c49
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
d079446
 
 
 
 
 
c109c49
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
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()